Hi Biao, Thanks for pointing out the inconsistency on TM management. I've corrected it. The taskmanagers are expected to be created by the resource manager driver, just like the native K8s mode.
As for artifact distribution, for the initial implementation, I propose supporting two options: JARs pre-provisioned in the execution environment, referenced through local:///..., and JARs already available through a configured Flink filesystem. The Ray application entrypoint would reuse Flink’s artifact-fetching infrastructure to resolve the application JAR and additional JARs specified through user.artifacts.artifact-list before loading the application. Additional JARs would be made available to the application classloader and explicitly registered as job dependencies for distribution to TaskManagers through Flink’s existing Blob services. Dependencies required in Flink’s lib/ or plugins/ directories, including filesystem plugins needed for artifact fetching, would remain prerequisites of the execution environment. To keep the initial scope bounded, client-local artifacts would need to be uploaded or pre-provisioned by the user. Automatic client-side upload and a dedicated integration with Ray’s runtime_env.working_dir would be deferred. Best, Zhanghao Chen ________________________________ From: Biao Geng <[email protected]> Sent: Wednesday, October 14, 2026 16:37 To: [email protected] <[email protected]> Subject: Re: [DISCUSS] FLIP-608: Flink on Ray, Ray Resource Backend For Flink Hi Zhanghao, Thanks for proposing this FLIP! Integrating Flink with Ray would help Flink connect more closely with the Ray and Python ecosystems, making its stream processing capabilities more accessible to Python and AI users. The overall architecture looks reasonable to me as well and follows a similar approach to Flink’s existing YARN and Kubernetes integrations. I have two follow-up questions: *1. TaskManager actor creation* The “Resource Management” section describes RayResourceManagerDriver requesting TaskManager actors from Ray. However, “Startup and Runtime Flow” says that the Python driver or submission entry point creates both the JobManager actor and a number of TaskManager actors. Is the intended design that the submission entry point creates only the JobManager actor, and RayResourceManagerDriver requests all TaskManager actors, including the initial ones? This would align with native Flink on Kubernetes. It would help to make these two sections consistent about who is responsible for TaskManager provisioning. *2. Application artifact and dependency distribution* Could the FLIP clarify how application artifacts will be provided when using the ray-application deployment target? Native Flink on Kubernetes supports application JARs bundled in the container image, as well as uploading local artifacts to remote storage and fetching them during cluster startup. Will the Ray backend reuse this approach or leverage Ray’s file distribution mechanisms such as runtime_env.working_dir? It would help to document the initial deployment requirements for application JARs and additional connector JARs, and how these integrate with Flink’s existing artifact management.Best, Biao Geng Dian Fu <[email protected]> 于2026年10月8日周四 13:58写道： > Hi Zhanghao, > > Thanks for driving this FLIP. This is an interesting topic. > > Ray has become an important execution layer in the Python data and AI > ecosystem. Besides Ray-native libraries, there are quite a few engines > such as Daft and Modin that use Ray as an execution backend, and Dask > workloads can also run through Dask-on-Ray. Supporting Flink on Ray > would allow PyFlink users to reuse an existing Python-native cluster > and share Ray’s CPU, GPU, and resource scheduling capabilities. > > The overall architecture looks reasonable: use Ray actors as > deployment units that supervise Flink JVM processes, while relying on > Ray for resource allocation and keeping scheduling, checkpointing, and > task recovery in the existing Flink runtime. > > I have several questions about the design details. > > ### TaskManager actor creation and teardown > > The FLIP could further clarify how TaskManager actor creation and > ownership work across a JobManager failover, for example: > > - Whether a JobManager failover causes existing TaskManager actors to > restart. My expectation is that healthy TaskManager actors and their > JVM processes should remain running. > - How a recovered JobManager obtains handles to existing TaskManager > actors. > - How duplicate TaskManager actor creation is prevented during failover. > - Whether TaskManager actors are named (if so, how the name will be > derived) and detached from their creator. > - How stale worker-management requests from the previous JobManager > are fenced after a failover? > - How detached TaskManager actors are cleaned up after the application > reaches a terminal state. > > The design should also distinguish a JobManager restart from an > application shutdown. Restarting or replacing the JobManager should > preserve healthy TaskManager actors, while application shutdown should > explicitly and reliably terminate all detached TaskManager actors. > > ### High availability model > > Whether the proposed deployment reuses an existing Flink HA > implementation or will introduce a new `RayHighAvailabilityServices`? > > Besides, whether JobManager HA supports the following cases: > - one restartable JobManager actor; or > - multiple active/standby JobManager actors. > > ### GPU resource management > > The FLIP could clarify the GPU allocation and isolation granularity. > > When a TaskManager actor requests GPU resources, Ray sets > `CUDA_VISIBLE_DEVICES` for the actor. The TaskManager JVM and the > Python UDF workers launched by it may therefore see the same set of > GPUs. If multiple GPU-consuming Python workers run concurrently in the > same TaskManager, they may compete for GPU memory and compute > resources. > > Configuring one slot per GPU-enabled TaskManager can reduce the > likelihood of such contention, but it does not by itself guarantee > isolation. A Flink slot may contain multiple operators or operator > chains, and an operator chain may contain multiple GPU-consuming UDFs. > > Supporting multiple independently isolated GPU consumers in one > TaskManager would require an additional per-slot or per-Python-worker > device allocation mechanism, including assigning a distinct > `CUDA_VISIBLE_DEVICES` value when each Python worker is launched. The > behavior should be clearly defined. For the initial version, the FLIP > could also document that GPU allocation and visibility are defined at > the TaskManager actor level and that no additional slot- or UDF-level > isolation is provided. > > ### API and naming > > The FLIP currently uses both `init_flink()` and `init_flink_on_ray()`. > I guess it could be simplified as following: > > import pyflink.ray as flink_on_ray > flink_on_ray.init(...) > > The FLIP could also clarify whether `init()` invokes `ray.init()` > automatically, how it behaves when a Ray context already exists, and > whether the corresponding shutdown operation terminates only the Flink > application or the entire Ray cluster. > > The actor names could also be made consistent. What about renaming > `RayFlinkJobManagerActor` to `RayJobManagerActor` so that it is > symmetric with `RayTaskManagerActor`. > > Thanks, > Dian > > On Fri, Sep 18, 2026 at 12:50 PM Zhanghao Chen > <[email protected]> wrote: > > > > Hi all, > > I would like to start a discussion about FLIP-608: Flink on Ray — Ray > Resource Backend For Flink [1]. > > This FLIP proposes a new Ray resource backend to allow running Flink > jobs natively on Ray. The hybrid approach takes the best of both worlds > —battle-tested streaming and batch unific computing plus heterogeneous > scheduling and the Ray AI eco-system (Data/Serve/Train). One can therefore > setup the entire real-time AI workflow in a single cluster with higher > resource efficiency and lower maintenance cost. > > The proposal focuses on supporting running Flink jobs on Ray in > Application/Session mode by translating Flink resource requirements to Ray. > The Flink core mechanisms like RPC, shuffle, checkpointing, failover and > slot management are preserved. > > A follow-up FLIP will focus on the interoperability between PyFlink and > Ray by supporting conversion between Flink Dataframe/Datastream/Table and > Ray Dataset. > > Looking forward to your feedback! > > [1] > https://cwiki.apache.org/confluence/spaces/FLINK/pages/449286348/FLIP-608+Flink+on+Ray+%E2%80%94+Ray+Resource+Backend+For+Flink > > > > Best, > > Zhanghao Chen >
