[Data] OpTask._cancel never passes force=True

[Hyunoh-Yeo]

Description

Current behavior of OpTask._cancel never passes force=True to ray.cancel. It is because whether the task is an actor task is determined by checking if the actor id is nil, which is false for both normal tasks and actor tasks (refer to the issue).

Slicing the hex method was withdrawn through discussions with maintainers. Instead, removed the check and lets Ray Core classify the task, with a fallback to force=False when Core rejects it for an actor task.

Related issues

Closes #65280

Additional information

Two tests added in TestOpTaskCancel
(python/ray/data/tests/test_streaming_executor.py)

• One test for a normal task
• One test for an actor task

[gemini-code-assist]

Code Review

This pull request refactors the task cancellation logic in physical_operator.py to use a try-except block when calling ray.cancel, falling back to force=False if a ValueError is raised. Unit tests were also added to verify this behavior. The review feedback suggests optimizing this logic to avoid redundant calls to ray.cancel when force is already False.

[gemini-code-assist]

Code Review

This pull request refactors the task cancellation logic in physical_operator.py to use a try-except block instead of checking if the task is an actor task beforehand. If a ValueError is raised during a force-cancellation attempt, it falls back to a non-forced cancellation. Unit tests are also added to verify this behavior. The reviewer suggested optimizing the cancellation logic to avoid a redundant call to ray.cancel when force is already False and a ValueError is raised.

[gemini-code-assist]

If force is False and ray.cancel raises a ValueError for any other reason, the current implementation will catch it and redundantly call ray.cancel with force=False again.

We can simplify this logic and avoid the redundant call by only wrapping the force=True call in the try-except block.

Suggested change

	try:
	ray.cancel(waitable, recursive=True, force=force)
	except ValueError:
	# Actor tasks can't be force cancelled.
	# If the task is an actor task, fallback to force=False.
	ray.cancel(waitable, recursive=True, force=False)
	if force:
	try:
	ray.cancel(waitable, recursive=True, force=True)
	return
	except ValueError:
	# Actor tasks can't be force cancelled.
	# If the task is an actor task, fallback to force=False.
	pass
	ray.cancel(waitable, recursive=True, force=False)

[Hyunoh-Yeo]

Even though force is passed to _cancel, conditioning the value of force with if/else and rewriting ray.cancel looks even more redundant. Sticking with the current, clearer option. Also, ValueError does not occur when force is False.

[bveeramani]

Nit: Rather than checking that we first call with force and then without, can we just check that we call with force=False at least once?

If we change the implementation to explicitly check if a waitable is for an actor task and decide whether to use force accordingly, then this will test will fail