[data] Failed to submit task to actor after ray.shutdown() and re-ray.init() in data pipeline for an existing cluster

[praateekmahajan]

What happened + What you expected to happen

If we have an existing ray cluster (e.g. setup using cli ray start --num-cpus 64) and we run two Ray Data pipelines back to back - each pipeline using ray.init(), processing a dataset with map_batches, and followed by ray.shutdown() - the second pipeline fails with an error indicating Ray cannot submit a task to an actor because it "can't find actor ... it might be dead or it's from a different cluster."

The use case is that we want a cluster to be instantiated independent of our pipelines, and each of our pipelines might be executed one after the other (with possibly different environment variables, so you can imagine the second pipeline sepcifies runtime_env={"env_vars": {"XYZ": "VALUE"}} while the first doesn't)

Versions / Dependencies

('2.48.0', '03491225d59a1ffde99c3628969ccf456be13efd')

Reproduction script

import os

import ray

# start a ray cluster using `ray start`
os.environ["RAY_ADDRESS"] = "127.0.0.1:8265"


class F:
    def __call__(self, x):
        return x

# First pipeline
ray.init(ignore_reinit_error=True)
ray.data.range(1000).map_batches(F, concurrency=(1, 4), num_cpus=1,).take_all()
ray.shutdown()

# Second pipeline
ray.init(ignore_reinit_error=True)
ray.data.range(1000).map_batches(F, concurrency=(1, 4), num_cpus=1,).take_all()
ray.shutdown()

Error

2025-07-22 21:34:22,645	ERROR exceptions.py:73 -- Exception occurred in Ray Data or Ray Core internal code. If you continue to see this error, please open an issue on the Ray project GitHub page with the full stack trace below: https://github.com/ray-project/ray/issues/new/choose
2025-07-22 21:34:22,645	ERROR exceptions.py:81 -- Full stack trace:
Traceback (most recent call last):
  File "/conda_env/lib/python3.12/site-packages/ray/data/exceptions.py", line 49, in handle_trace
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/data/_internal/plan.py", line 439, in execute_to_iterator
    bundle_iter = execute_to_legacy_bundle_iterator(executor, self)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/data/_internal/execution/legacy_compat.py", line 55, in execute_to_legacy_bundle_iterator
    bundle_iter = executor.execute(dag, initial_stats=stats)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/data/_internal/execution/streaming_executor.py", line 159, in execute
    StatsManager.register_dataset_to_stats_actor(
  File "/conda_env/lib/python3.12/site-packages/ray/data/_internal/stats.py", line 790, in register_dataset_to_stats_actor
    self._stats_actor().register_dataset.remote(
  File "/conda_env/lib/python3.12/site-packages/ray/actor.py", line 676, in remote
    return self._remote(args, kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/_private/auto_init_hook.py", line 22, in auto_init_wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/util/tracing/tracing_helper.py", line 422, in _start_span
    return method(self, args, kwargs, *_args, **_kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/actor.py", line 856, in _remote
    obj_ref = invocation(args, kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/actor.py", line 836, in invocation
    return dst_actor._actor_method_call(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/conda_env/lib/python3.12/site-packages/ray/actor.py", line 2059, in _actor_method_call
    object_refs = worker.core_worker.submit_actor_task(
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "python/ray/_raylet.pyx", line 3992, in ray._raylet.CoreWorker.submit_actor_task
  File "python/ray/_raylet.pyx", line 4053, in ray._raylet.CoreWorker.submit_actor_task
Exception: Failed to submit task to actor ActorID(4264d583dfdde28efb09616901000000) due to b"Can't find actor 4264d583dfdde28efb09616901000000. It might be dead or it's from a different cluster"

Issue Severity

Medium: It is a significant difficulty but I can work around it.

[praateekmahajan]

Looking at the traceback and then the code it seems like the StatsActor is pinned to the driver node, so even though ray shutdown kills the actor, the subsequent ray.data pipeline tries to use it again but that actor doesn't exist anymore

ray/python/ray/data/_internal/stats.py

Lines 576 to 593 in 753a275

	def _get_or_create_stats_actor():
	ctx = DataContext.get_current()
	scheduling_strategy = ctx.scheduling_strategy
	if not ray.util.client.ray.is_connected():
	# Pin the stats actor to the local node
	# so it fate-shares with the driver.
	scheduling_strategy = NodeAffinitySchedulingStrategy(
	ray.get_runtime_context().get_node_id(),
	soft=False,
	)
	with _stats_actor_lock:
	return _StatsActor.options(
	name=STATS_ACTOR_NAME,
	namespace=STATS_ACTOR_NAMESPACE,
	get_if_exists=True,
	lifetime="detached",
	scheduling_strategy=scheduling_strategy,
	).remote()

[alexeykudinkin]

@praateekmahajan can you help me understand why do you need to call ray.init and shutdown in your example b/w Ray Data executions?

[praateekmahajan]

The primary use case for us is that we do something like this

1. ray start .....
2. run_pipeline_one(); run_pipeline_two()
3. ray stop ...

• run_pipeline_one() might use ray data or might use our other execution backend called Xenna.
• run_pipeline_two() similarly might use ray data or something else.

Either of run_pipeline_one() might need certain environment variables, which we do using ray.init(ignore_reinit=True, runtime_env={envvars={...}}). And you can't unset these unless you have performed a ray.shutdown(..). This is the need for having ray.init(...) and ray.shutdown(), and then in theory a user might run two both run_pipeline_one() run_pipeline_two() and can be ray data pipelines.

As a sidenote, we'd love something like ray.disconnect() which disasocciates any actors / envvars created in the current context, instead of ray.shutdown. And in this case it unpins the StatsActor option

[alexeykudinkin]

and then in theory a user might run two both run_pipeline_one() run_pipeline_two() and can be ray data pipelines.

Can you give examples of such pipelines? I'm trying to understand why we need to run them inside the same driver (ie script calling ray.init) and therefore being wrapped around with init/shutdown.

If you separate them out into different drivers this will be handled for you automatically.

[alexeykudinkin]

@praateekmahajan btw, when you call ray.shutdown you actually don't shutdown the Ray cluster, you just disconnect the driver from it.

If you want to shutdown the cluster (that was started with ray start) you'd need to call ray stop