[hotfix][coordination] Guard assumption in AdaptiveScheduler.Restarting

is · Mar 15, 2021 · e8c146e · e8c146e
1 parent 18a9f02
commit e8c146e
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
@@ -23,6 +23,7 @@
 import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
 import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
 import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
+import org.apache.flink.util.Preconditions;
 
 import org.slf4j.Logger;
 
@@ -72,9 +73,8 @@ boolean updateTaskExecutionState(TaskExecutionStateTransition taskExecutionState
 
  @Override
  void onGloballyTerminalState(JobStatus globallyTerminalState) {
- if (globallyTerminalState == JobStatus.CANCELED) {
- context.runIfState(this, context::goToWaitingForResources, backoffTime);
- }
+ Preconditions.checkArgument(globallyTerminalState == JobStatus.CANCELED);
+ context.runIfState(this, context::goToWaitingForResources, backoffTime);
  }
 
  /** Context of the {@link Restarting} state. */