barakbaron12 commented on PR #27334: URL: https://github.com/apache/flink/pull/27334#issuecomment-4194087336
Thanks for the follow-up @DongyuanPan — I think I understand now. If I'm reading you correctly, the idea is to merge the cluster-side overrides into the StreamGraph/JobGraph's own configuration inside `internalSubmitJob` (since the Dispatcher has access to both), so that all downstream code only needs to read from a single config source. Let me know if I'm misunderstanding. I see the appeal — it would centralize the merge logic in one place. I do have a couple of concerns though: 1. Recovery path: `runRecoveredJob` calls `createJobMasterRunner` directly without going through `internalSubmitJob`. So if a job recovers after a failover, the cluster-side overrides wouldn't get merged into the plan's config. We'd either need to duplicate the merge logic in the recovery path (which defeats the single-place benefit), or accept that recovered jobs use whatever overrides were baked in at original submission time. With the current approach, the scheduler factories always merge fresh cluster config on every run — whether it's a new submission or a recovery. 2. Mutating the persisted plan: `persistAndRunJob` writes the execution plan to storage before running it. If we merge cluster overrides into the plan's config first, the persisted version includes those operational overrides. This means you can no longer tell from the persisted plan which overrides came from the job author vs. the cluster operator. Keeping them separate until the scheduler merges them preserves that distinction. 3. Adaptive batch complexity: As you noted, the `AdaptiveBatchScheduler` builds the JobGraph incrementally — vertices are created one-by-one as upstream tasks finish. The override check needs to happen per-vertex at decision time (in `getInitialParallelism()`), because that's when the scheduler decides between "use the override" vs "decide parallelism adaptively based on upstream data volume." This per-vertex logic stays the same regardless of where the config merge happens, so moving it to the Dispatcher doesn't simplify the adaptive path — which is really the trickiest part of this change. 4. The downstream logic stays the same either way: The scheduler factories still need to call `applyParallelismOverrides()` to set vertex parallelism, and `DefaultAdaptiveExecutionHandler.getInitialParallelism()` still needs the per-vertex override check for the adaptive path. The only difference is whether they read from one source or merge two — the application logic doesn't change. Happy to discuss further if you see a way around these, or if I'm misunderstanding your suggestion! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
