zhuzhurk commented on code in PR #24223:
URL: https://github.com/apache/flink/pull/24223#discussion_r1472738431
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java:
##########
@@ -187,17 +189,28 @@ public AdaptiveBatchScheduler(
protected void startSchedulingInternal() {
tryComputeSourceParallelismThenRunAsync(
(Void value, Throwable throwable) -> {
- initializeVerticesIfPossible();
- super.startSchedulingInternal();
+ if (getExecutionGraph().getState() == JobStatus.CREATED) {
+ initializeVerticesIfPossible();
+ super.startSchedulingInternal();
+ }
});
}
@Override
protected void onTaskFinished(final Execution execution, final IOMetrics
ioMetrics) {
checkNotNull(ioMetrics);
updateResultPartitionBytesMetrics(ioMetrics.getResultPartitionBytes());
+ ExecutionVertexVersion currentVersion =
+
executionVertexVersioner.getExecutionVertexVersion(execution.getVertex().getID());
tryComputeSourceParallelismThenRunAsync(
(Void value, Throwable throwable) -> {
+ if (executionVertexVersioner.isModified(currentVersion)) {
+ log.debug(
+ "Initialization of vertices will be skipped,
because the execution"
+ + " vertex version has been modified,
suggesting that a global "
Review Comment:
I tend to omit the last part "suggesting ..." because it may also happen in
task failover cases. e.g. if B1 fails due to `PartitionException`, the upstream
tasks will be restarted and B2 is possible to be restarted too, even if it has
finished.
--
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]
