zhuzhurk commented on a change in pull request #9663:
[WIP][FLINK-12433][runtime] Implement DefaultScheduler stub
URL: https://github.com/apache/flink/pull/9663#discussion_r326897974
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
##########
@@ -75,10 +137,281 @@ public DefaultScheduler(
slotRequestTimeout,
shuffleMaster,
partitionTracker);
+
+ this.restartBackoffTimeStrategy =
checkNotNull(restartBackoffTimeStrategy);
+ this.slotRequestTimeout = checkNotNull(slotRequestTimeout);
+ this.slotProvider = checkNotNull(slotProvider);
+ this.delayExecutor = checkNotNull(delayExecutor);
+ this.userCodeLoader = checkNotNull(userCodeLoader);
+ this.schedulingStrategyFactory =
checkNotNull(schedulingStrategyFactory);
+ this.failoverStrategyFactory =
checkNotNull(failoverStrategyFactory);
+ this.executionVertexOperations =
checkNotNull(executionVertexOperations);
+ this.executionVertexVersioner =
checkNotNull(executionVertexVersioner);
+ this.conditionalFutureHandlerFactory = new
ConditionalFutureHandlerFactory(executionVertexVersioner);
}
+ //
------------------------------------------------------------------------
+ // SchedulerNG
+ //
------------------------------------------------------------------------
+
@Override
- public void startScheduling() {
- throw new UnsupportedOperationException();
+ public void startSchedulingInternal() {
+ initializeScheduling();
+ schedulingStrategy.startScheduling();
+ }
+
+ private void initializeScheduling() {
+ executionFailureHandler = new
ExecutionFailureHandler(failoverStrategyFactory.create(getFailoverTopology()),
restartBackoffTimeStrategy);
+ schedulingStrategy =
schedulingStrategyFactory.createInstance(this, getSchedulingTopology(),
getJobGraph());
+ executionSlotAllocator = new
DefaultExecutionSlotAllocator(slotProvider, getInputsLocationsRetriever(),
slotRequestTimeout);
+ setTaskFailureListener(new
UpdateTaskExecutionStateInDefaultSchedulerListener(this,
getJobGraph().getJobID()));
+ prepareExecutionGraphForScheduling();
+ }
+
+ @Override
+ public boolean updateTaskExecutionState(final TaskExecutionState
taskExecutionState) {
+ final Optional<ExecutionVertexID> executionVertexIdOptional =
getExecutionVertexId(taskExecutionState.getID());
+ if (executionVertexIdOptional.isPresent()) {
+ final ExecutionVertexID executionVertexId =
executionVertexIdOptional.get();
+ updateState(taskExecutionState);
+
schedulingStrategy.onExecutionStateChange(executionVertexId,
taskExecutionState.getExecutionState());
Review comment:
By design execution state changes should be notified via `SchedulingStrategy
#onExecutionStateChange`.
Similarly, all partition consumable state changes should be notified via
`SchedulingStrategy #onPartitionConsumable`, as discussed the in [another
comment](https://github.com/apache/flink/pull/9663/files#r326540913)
This is how we define this public interface and I think we need to guarantee
its correctness eventually.
And you are right that it is not a problem at the moment, or before we mark
it as a public interface for users or other developers to customize the
scheduling strategy.
So I think we can skip it in this PR and do this in a follow up task.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
