XComp commented on code in PR #27334:
URL: https://github.com/apache/flink/pull/27334#discussion_r3566946198
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultAdaptiveExecutionHandler.java:
##########
@@ -69,6 +78,14 @@ public DefaultAdaptiveExecutionHandler(
new StreamGraphOptimizer(streamGraph.getJobConfiguration(),
userClassloader);
this.streamGraphOptimizer.initializeStrategies(
adaptiveGraphManager.getStreamGraphContext());
+
+ // Merge overrides: master config first, job config wins (same
precedence as
Review Comment:
We're skipping the validation of the overrides here which would happen in
`ParallelismOverrideUtil#parseParallelism§` I guess that should be fixed before
merging, hm?
##########
flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java:
##########
@@ -404,4 +413,107 @@ public boolean onOperatorsFinished(
return context.modifyStreamEdge(requestInfos);
}
}
+
+ /**
+ * Tests that parallelism overrides work correctly with the AdaptiveBatch
scheduler. This
Review Comment:
```suggestion
* Tests that parallelism overrides work correctly with the {@code
AdaptiveBatchScheduler}. This
```
nit
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java:
##########
@@ -122,7 +123,10 @@ public SchedulerNG createInstance(
Collection<FailureEnricher> failureEnrichers,
BlocklistOperations blocklistOperations)
throws Exception {
- ExecutionConfig executionConfig;
+ final ExecutionConfig executionConfig;
+
+ ParallelismOverrideUtil.applyParallelismOverridesIfApplicable(
+ executionPlan, jobMasterConfiguration);
Review Comment:
Wouldn't it make sense to move this also into
`AdaptiveExecutionHandlerFactory`. I mean we would still have the split of
responsibilities between `AdaptiveExecutionHandlerFactory` and
`DefaultAdaptiveExecutionHandler`. But we would at least make it a bit more
consistent. WDYT?
We could delete
`ParallelismOverrideUtil#applyParallelismOverridesIfApplicable` then.
##########
flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java:
##########
@@ -404,4 +413,107 @@ public boolean onOperatorsFinished(
return context.modifyStreamEdge(requestInfos);
}
}
+
+ /**
+ * Tests that parallelism overrides work correctly with the AdaptiveBatch
scheduler. This
+ * verifies the fix for FLINK-38770.
+ */
+ @Test
+ void testParallelismOverridesWithAdaptiveBatchScheduler() throws Exception
{
+ Configuration configuration = createConfiguration();
+ StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ env.setParallelism(1);
+
+ // Create a simple batch job
+ env.fromSequence(0, 100).map(i -> i * 2).name("test-map").print();
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ JobGraph jobGraph = streamGraph.getJobGraph();
+
+ // Find the map vertex and configure parallelism override
+ JobVertex mapVertex = null;
+ for (JobVertex vertex : jobGraph.getVertices()) {
+ if (vertex.getName().contains("test-map")) {
+ mapVertex = vertex;
+ break;
+ }
+ }
+ assertThat(mapVertex).isNotNull();
+
+ // Configure parallelism override to change parallelism to 2
+ Map<String, String> overrides = new HashMap<>();
+ overrides.put(mapVertex.getID().toHexString(), "2");
+
jobGraph.getJobConfiguration().set(PipelineOptions.PARALLELISM_OVERRIDES,
overrides);
+
+ // Submit and run the job - it will use the overridden parallelism (2
slots needed)
+ JobGraphRunningUtil.execute(jobGraph, configuration, 1, 2);
Review Comment:
We have to find a better assert here: The test also succeeds if the override
is not applied. I guess that's due to the fact that the job will just executed
in multiple batches.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveExecutionHandlerFactory.java:
##########
@@ -54,18 +58,25 @@ public static AdaptiveExecutionHandler create(
ExecutionPlan executionPlan,
boolean enableBatchJobRecovery,
ClassLoader userClassLoader,
- Executor serializationExecutor)
+ Executor serializationExecutor,
+ Configuration jobMasterConfiguration)
throws DynamicCodeLoadingException {
if (executionPlan instanceof JobGraph) {
return new NonAdaptiveExecutionHandler((JobGraph) executionPlan);
} else {
checkState(executionPlan instanceof StreamGraph, "Unsupported
execution plan.");
if (enableBatchJobRecovery) {
StreamGraph streamGraph = (StreamGraph) executionPlan;
- return new
NonAdaptiveExecutionHandler(streamGraph.getJobGraph(userClassLoader));
+ JobGraph jobGraph = streamGraph.getJobGraph(userClassLoader);
+ // Apply parallelism overrides after StreamGraph -> JobGraph
conversion
+ ParallelismOverrideUtil.applyParallelismOverrides(jobGraph,
jobMasterConfiguration);
+ return new NonAdaptiveExecutionHandler(jobGraph);
} else {
return new DefaultAdaptiveExecutionHandler(
- userClassLoader, (StreamGraph) executionPlan,
serializationExecutor);
+ userClassLoader,
+ (StreamGraph) executionPlan,
+ serializationExecutor,
+ jobMasterConfiguration);
Review Comment:
Instead of passing in `jobMasterConfiguration`, we could also extract the
per-vertex parallelism overrides here and pass them into
`DefaultAdaptiveExecutionHandler`. That way (with the [other suggestion
above](https://github.com/apache/flink/pull/27334/changes#r3566978850)
considered), we could move the override handling entirely into
`AdaptiveExecutionHandlerFactory`.
--
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]
