[data] Add node_id, pid, attempt # for hanging tasks

[iamjustinhsu]

Description

Currently, when displaying hanging tasks, we show ray data level task index, which is useless for ray core debugging. This PR adds more info to long running tasks namely:

• node_id
• pid
• attempt #

I did consider adding this to high memory detector, but avoided for 2 reasons

• requires more refractor of RunningTaskInfo
• afaik, not helpful in debugging since high memory is after the task completes

Example script to trigger hanging issues

import ray
import time
from ray.data._internal.issue_detection.detectors import HangingExecutionIssueDetectorConfig


ctx = ray.data.DataContext.get_current()
ctx.issue_detectors_config.hanging_detector_config = HangingExecutionIssueDetectorConfig(
    detection_time_interval_s=1.0,
)

def sleep(x):
    if x['id'] == 0:
        time.sleep(100)
    return x
ray.data.range(100, override_num_blocks=100).map_batches(sleep).materialize()

Related issues

None

Additional information

None

[gemini-code-assist]

Code Review

This pull request aims to enhance the debuggability of hanging tasks by incorporating node_id, pid, and attempt # into the hanging task detector's output. This is achieved by passing the task_id through the operator pipeline to OpRuntimeMetrics, which is then used by the HangingExecutionIssueDetector to fetch detailed task information. The implementation is generally sound, with a beneficial refactoring in physical_operator.py. However, I've identified a critical issue in hash_shuffle.py where arguments to on_task_submitted are incorrectly ordered, which would result in a runtime error.

[goutamvenkat-anyscale]

I assume this state is not serialized and persisted anywhere correct?

[iamjustinhsu]

nah it's not

[goutamvenkat-anyscale]

Can we pass in _explain=True and log the explanation in the event of a failure?