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: us...@infra.apache.org With regards, Apache Git Services