Repository: ray-project/ray
File: python/ray/data/_internal/execution/interfaces/physical_operator.py (Python)

Code before:
                        self._has_finished = True
                        raise ex from None

                if self._pending_meta_ref.is_nil():
                    # We have a reference to the block but the metadata isn't ready
                    # yet.
                    break

                self._metadata_ready_callback(self._pending_meta_ref)

            try:
                # The timeout for `ray.get` includes the time required to ship the
                # block metadata to this node. So, if we set the timeout to 0, `ray.get`
                # will timeout and possible cancel the download. To avoid this issue,
                # we set the timeout to a small non-zero value.
                meta_with_schema: "BlockMetadataWithSchema" = ray.get(
                    self._pending_meta_ref, timeout=METADATA_GET_TIMEOUT_S
                )
            except ray.exceptions.GetTimeoutError:
                # We have a reference to the block and its metadata, but the metadata
                # object isn't available. This can happen if the node dies.
                operator_name = self._operator_name or self.__class__.__name__
                logger.warning(
                    f"Timed out ({METADATA_GET_TIMEOUT_S}s) waiting for "
                    f"operator '{operator_name}' (ref={self._pending_meta_ref.hex()}). "
                    f"Possible causes: slow task, worker crashed, or cluster overloaded. "
                    f"Will retry next iteration. "
                    f"If this repeats, check task/worker logs and cluster resources."
                )
                break

            meta = meta_with_schema.metadata
            self._output_ready_callback(
                RefBundle(
                    [(self._pending_block_ref, meta)],
                    owns_blocks=True,
                    schema=meta_with_schema.schema,
                ),
            )

            self._last_block_meta = meta
            self._pending_block_ref = ray.ObjectRef.nil()
            self._pending_meta_ref = ray.ObjectRef.nil()

            bytes_read += meta.size_bytes

        return bytes_read

    def _track_task_output_backpressure(self, max_bytes_to_read: Optional[int]):
        if max_bytes_to_read == 0:
            # Whenever provided `max_bytes_to_read == 0` we treat as task

Review comment (bug):
Don't think slow tasks can usually cause this (?)

IIUC it's usually because of worker crash/node preemption or overloaded worker/head node.

Code after:
                        self._has_finished = True
                        raise ex from None

                if self._pending_meta_ref.is_nil():
                    # We have a reference to the block but the metadata isn't ready
                    # yet.
                    break

                self._metadata_ready_callback(self._pending_meta_ref)

            try:
                # The timeout for `ray.get` includes the time required to ship the
                # block metadata to this node. So, if we set the timeout to 0, `ray.get`
                # will timeout and possible cancel the download. To avoid this issue,
                # we set the timeout to a small non-zero value.
                meta_with_schema: "BlockMetadataWithSchema" = ray.get(
                    self._pending_meta_ref, timeout=METADATA_GET_TIMEOUT_S
                )
            except ray.exceptions.GetTimeoutError:
                # We have a reference to the block and its metadata, but the metadata
                # object isn't available. This can happen if the node dies.
                logger.warning(
                    f"Timed out ({METADATA_GET_TIMEOUT_S}s) waiting for metadata from "
                    f"operator '{self._operator_name}' "
                    f"(metadata_ref={self._pending_meta_ref.hex()}). "
                    f"Possible causes include a worker crash, node preemption, or an overloaded worker or head node. "
                    f"Will retry next iteration. "
                    f"If this repeats, check the Ray dashboard and logs for worker crashes, node preemption, or overload."
                )
                break

            meta = meta_with_schema.metadata
            self._output_ready_callback(
                RefBundle(
                    [(self._pending_block_ref, meta)],
                    owns_blocks=True,
                    schema=meta_with_schema.schema,
                ),
            )

            self._last_block_meta = meta
            self._pending_block_ref = ray.ObjectRef.nil()
            self._pending_meta_ref = ray.ObjectRef.nil()

            bytes_read += meta.size_bytes

        return bytes_read

    def _track_task_output_backpressure(self, max_bytes_to_read: Optional[int]):
        if max_bytes_to_read == 0:
            # Whenever provided `max_bytes_to_read == 0` we treat as task