Hi Zhanghao, Thanks a lot for the update. It addresses most of my concerns.
A few inconsistencies in the FLIP: - `RayJobManagerActor` is used in the runtime interface, while `RayFlinkJobManager Actor` still appears in the component mapping, Py4J integration, and startup flow. It would be clearer to use `RayJobManagerActor` consistently. - The startup flow and runtime interface use `pyflink.ray.init()` / `init()`, but the example still calls `flinkray.init_flink()`. Regards, Dian On Sat, Oct 17, 2026 at 1:12 AM Zhanghao Chen <[email protected]> wrote: > > 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 > >
