[FLINK-22215][runtime] RECOVERING state is renamed to INITIALIZING

deepfield · Apr 13, 2021 · b387928 · b387928
1 parent f829418
commit b387928
Show file tree

Hide file tree

Showing 23 changed files with 72 additions and 72 deletions.
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -1558,7 +1558,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "tasks" : {
  "type" : "object",
@@ -3560,7 +3560,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "subtask" : {
  "type" : "integer"
@@ -4145,7 +4145,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "subtask" : {
  "type" : "integer"
@@ -4273,7 +4273,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "subtask" : {
  "type" : "integer"
@@ -4682,7 +4682,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "status-counts" : {
  "type" : "object",

diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -740,7 +740,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "start-time" : {
  "type" : "integer"
@@ -1967,7 +1967,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "attempt" : {
  "type" : "integer"
@@ -2279,7 +2279,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "attempt" : {
  "type" : "integer"
@@ -2365,7 +2365,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "attempt" : {
  "type" : "integer"
@@ -2600,7 +2600,7 @@
  },
  "status" : {
  "type" : "string",
- "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "RECOVERING" ]
+ "enum" : [ "CREATED", "SCHEDULED", "DEPLOYING", "RUNNING", "FINISHED", "CANCELING", "CANCELED", "FAILED", "RECONCILING", "INITIALIZING" ]
  },
  "start-time" : {
  "type" : "integer"

diff --git a/flink-runtime-web/web-dashboard/src/app/app.config.ts b/flink-runtime-web/web-dashboard/src/app/app.config.ts
@@ -31,7 +31,6 @@ export const COLOR_MAP = {
  SCHEDULED: '#722ed1',
  COMPLETED: '#1890ff',
  RESTARTING: '#13c2c2',
- INITIALIZING: '#738df8',
- RECOVERING: '#18c98b'
+ INITIALIZING: '#738df8'
 };
 export const LONG_MIN_VALUE = -9223372036854776000;
diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-detail.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-detail.ts
@@ -104,7 +104,7 @@ interface TasksStatus {
  FAILED: number;
  RECONCILING: number;
  CANCELING: number;
- RECOVERING: number;
+ INITIALIZING: number;
 }
 
 interface MetricsStatus {

diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-overview.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-overview.ts
@@ -43,5 +43,5 @@ export interface TaskStatusInterface {
  RUNNING: number;
  SCHEDULED: number;
  TOTAL: number;
- RECOVERING: number;
+ INITIALIZING: number;
 }
diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-vertex-task-manager.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-vertex-task-manager.ts
@@ -49,7 +49,7 @@ export interface VertexTaskManagerDetailInterface {
  RECONCILING: number;
  RUNNING: number;
  SCHEDULED: number;
- RECOVERING: number;
+ INITIALIZING: number;
  };
  'taskmanager-id': string;
 }
diff --git a/...me/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java b/...me/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java
@@ -258,7 +258,7 @@ private static RuntimeException handleConsumedPartitionShuffleDescriptorErrors(
 
  private static boolean isProducerAvailable(ExecutionState producerState) {
  return producerState == ExecutionState.RUNNING
- || producerState == ExecutionState.RECOVERING
+ || producerState == ExecutionState.INITIALIZING
  || producerState == ExecutionState.FINISHED
  || producerState == ExecutionState.SCHEDULED
  || producerState == ExecutionState.DEPLOYING;

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/ExecutionState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/ExecutionState.java
@@ -23,7 +23,7 @@
  * the state {@code CREATED} and switch states according to this diagram:
  *
  * <pre>{@code
- * CREATED -> SCHEDULED -> DEPLOYING -> RECOVERING -> RUNNING -> FINISHED
+ * CREATED -> SCHEDULED -> DEPLOYING -> INITIALIZING -> RUNNING -> FINISHED
  * | | | | |
  * | | | +-----+--------------+
  * | | V V
@@ -33,7 +33,7 @@
  * |
  * | ... -> FAILED
  * V
- * RECONCILING -> RECOVERING | RUNNING | FINISHED | CANCELED | FAILED
+ * RECONCILING -> INITIALIZING | RUNNING | FINISHED | CANCELED | FAILED
  *
  * }</pre>
  *
@@ -71,7 +71,7 @@ public enum ExecutionState {
  RECONCILING,
 
  /** Restoring last possible valid state of the task if it has it. */
- RECOVERING;
+ INITIALIZING;
 
  public boolean isTerminal() {
  return this == FINISHED || this == CANCELED || this == FAILED;

diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java b/...-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -957,7 +957,7 @@ void failGlobalIfExecutionIsStillRunning(Throwable cause, ExecutionAttemptID fai
  final Execution failedExecution = currentExecutions.get(failingAttempt);
  if (failedExecution != null
  && (failedExecution.getState() == ExecutionState.RUNNING
- || failedExecution.getState() == ExecutionState.RECOVERING)) {
+ || failedExecution.getState() == ExecutionState.INITIALIZING)) {
  failGlobal(cause);
  } else {
  LOG.debug(
@@ -1226,7 +1226,7 @@ private boolean updateStateInternal(
  Map<String, Accumulator<?, ?>> accumulators;
 
  switch (state.getExecutionState()) {
- case RECOVERING:
+ case INITIALIZING:
  return attempt.switchToRecovering();
 
  case RUNNING:

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -86,7 +86,7 @@
 import static org.apache.flink.runtime.execution.ExecutionState.DEPLOYING;
 import static org.apache.flink.runtime.execution.ExecutionState.FAILED;
 import static org.apache.flink.runtime.execution.ExecutionState.FINISHED;
-import static org.apache.flink.runtime.execution.ExecutionState.RECOVERING;
+import static org.apache.flink.runtime.execution.ExecutionState.INITIALIZING;
 import static org.apache.flink.runtime.execution.ExecutionState.RUNNING;
 import static org.apache.flink.runtime.execution.ExecutionState.SCHEDULED;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -623,7 +623,7 @@ public void cancel() {
  }
 
  // these two are the common cases where we need to send a cancel call
- else if (current == RECOVERING || current == RUNNING || current == DEPLOYING) {
+ else if (current == INITIALIZING || current == RUNNING || current == DEPLOYING) {
  // try to transition to canceling, if successful, send the cancel call
  if (startCancelling(NUM_CANCEL_CALL_TRIES)) {
  return;
@@ -663,7 +663,7 @@ else if (current == RECOVERING || current == RUNNING || current == DEPLOYING) {
  public CompletableFuture<?> suspend() {
  switch (state) {
  case RUNNING:
- case RECOVERING:
+ case INITIALIZING:
  case DEPLOYING:
  case CREATED:
  case SCHEDULED:
@@ -723,7 +723,7 @@ private void updatePartitionConsumers(final IntermediateResultPartition partitio
  // ----------------------------------------------------------------
  if (consumerState == DEPLOYING
  || consumerState == RUNNING
- || consumerState == RECOVERING) {
+ || consumerState == INITIALIZING) {
  final PartitionInfo partitionInfo = createPartitionInfo(partition);
 
  if (consumerState == DEPLOYING) {
@@ -856,7 +856,7 @@ public CompletableFuture<Acknowledge> sendOperatorEvent(
  OperatorID operatorId, SerializedValue<OperatorEvent> event) {
  final LogicalSlot slot = assignedResource;
 
- if (slot != null && (getState() == RUNNING || getState() == RECOVERING)) {
+ if (slot != null && (getState() == RUNNING || getState() == INITIALIZING)) {
  final TaskExecutorOperatorEventGateway eventGateway = slot.getTaskManagerGateway();
  return eventGateway.sendOperatorEventToTask(getAttemptId(), operatorId, event);
  } else {
@@ -907,7 +907,7 @@ void markFinished(Map<String, Accumulator<?, ?>> userAccumulators, IOMetrics met
  while (true) {
  ExecutionState current = this.state;
 
- if (current == RECOVERING || current == RUNNING || current == DEPLOYING) {
+ if (current == INITIALIZING || current == RUNNING || current == DEPLOYING) {
 
  if (transitionState(current, FINISHED)) {
  try {
@@ -999,7 +999,7 @@ void completeCancelling(
  return;
  } else if (current == CANCELING
  || current == RUNNING
- || current == RECOVERING
+ || current == INITIALIZING
  || current == DEPLOYING) {
 
  updateAccumulatorsAndMetrics(userAccumulators, metrics);
@@ -1138,7 +1138,7 @@ private void maybeReleasePartitionsAndSendCancelRpcCall(
 
  if (cancelTask
  && (stateBeforeFailed == RUNNING
- || stateBeforeFailed == RECOVERING
+ || stateBeforeFailed == INITIALIZING
  || stateBeforeFailed == DEPLOYING)) {
  if (LOG.isDebugEnabled()) {
  LOG.debug("Sending out cancel request, to remove task execution from TaskManager.");
@@ -1159,7 +1159,7 @@ private void maybeReleasePartitionsAndSendCancelRpcCall(
  }
 
  boolean switchToRecovering() {
- if (switchTo(DEPLOYING, RECOVERING)) {
+ if (switchTo(DEPLOYING, INITIALIZING)) {
  sendPartitionInfos();
  return true;
  }
@@ -1168,7 +1168,7 @@ boolean switchToRecovering() {
  }
 
  boolean switchToRunning() {
- return switchTo(RECOVERING, RUNNING);
+ return switchTo(INITIALIZING, RUNNING);
  }
 
  private boolean switchTo(ExecutionState from, ExecutionState to) {

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
@@ -521,8 +521,8 @@ public static ExecutionState getAggregateJobVertexState(
  return ExecutionState.CANCELING;
  } else if (verticesPerState[ExecutionState.CANCELED.ordinal()] > 0) {
  return ExecutionState.CANCELED;
- } else if (verticesPerState[ExecutionState.RECOVERING.ordinal()] > 0) {
- return ExecutionState.RECOVERING;
+ } else if (verticesPerState[ExecutionState.INITIALIZING.ordinal()] > 0) {
+ return ExecutionState.INITIALIZING;
  } else if (verticesPerState[ExecutionState.RUNNING.ordinal()] > 0) {
  return ExecutionState.RUNNING;
  } else if (verticesPerState[ExecutionState.FINISHED.ordinal()] > 0) {

diff --git a/...ava/org/apache/flink/runtime/io/network/partition/consumer/RemoteChannelStateChecker.java b/...ava/org/apache/flink/runtime/io/network/partition/consumer/RemoteChannelStateChecker.java
@@ -67,15 +67,15 @@ public boolean isProducerReadyOrAbortConsumption(ResponseHandle responseHandle)
  private static boolean isConsumerStateValidForConsumption(
  ExecutionState consumerExecutionState) {
  return consumerExecutionState == ExecutionState.RUNNING
- || consumerExecutionState == ExecutionState.RECOVERING
+ || consumerExecutionState == ExecutionState.INITIALIZING
  || consumerExecutionState == ExecutionState.DEPLOYING;
  }
 
  private boolean isProducerConsumerReady(ResponseHandle responseHandle) {
  ExecutionState producerState = getProducerState(responseHandle);
  return producerState == ExecutionState.SCHEDULED
  || producerState == ExecutionState.DEPLOYING
- || producerState == ExecutionState.RECOVERING
+ || producerState == ExecutionState.INITIALIZING
  || producerState == ExecutionState.RUNNING
  || producerState == ExecutionState.FINISHED;
  }

diff --git a/...e/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java b/...e/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
@@ -112,7 +112,7 @@ public void deliverOperatorEventToCoordinator(
  final Execution exec = executionGraph.getRegisteredExecutions().get(taskExecutionId);
  if (exec == null
  || exec.getState() != ExecutionState.RUNNING
- && exec.getState() != ExecutionState.RECOVERING) {
+ && exec.getState() != ExecutionState.INITIALIZING) {
  // This situation is common when cancellation happens, or when the task failed while the
  // event was just being dispatched asynchronously on the TM side.
  // It should be fine in those expected situations to just ignore this event, but, to be

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -505,7 +505,7 @@ AbstractInvokable getInvokable() {
  public boolean isBackPressured() {
  if (invokable == null
  || consumableNotifyingPartitionWriters.length == 0
- || (executionState != ExecutionState.RECOVERING
+ || (executionState != ExecutionState.INITIALIZING
  && executionState != ExecutionState.RUNNING)) {
  return false;
  }
@@ -738,14 +738,14 @@ private void doRun() {
  // by the time we switched to running.
  this.invokable = invokable;
 
- // switch to the RECOVERING state, if that fails, we have been canceled/failed in the
+ // switch to the INITIALIZING state, if that fails, we have been canceled/failed in the
  // meantime
- if (!transitionState(ExecutionState.DEPLOYING, ExecutionState.RECOVERING)) {
+ if (!transitionState(ExecutionState.DEPLOYING, ExecutionState.INITIALIZING)) {
  throw new CancelTaskException();
  }
 
  taskManagerActions.updateTaskExecutionState(
- new TaskExecutionState(executionId, ExecutionState.RECOVERING));
+ new TaskExecutionState(executionId, ExecutionState.INITIALIZING));
 
  // make sure the user code classloader is accessible thread-locally
  executingThread.setContextClassLoader(userCodeClassLoader.asClassLoader());
@@ -758,7 +758,7 @@ private void doRun() {
  FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
  }
 
- if (!transitionState(ExecutionState.RECOVERING, ExecutionState.RUNNING)) {
+ if (!transitionState(ExecutionState.INITIALIZING, ExecutionState.RUNNING)) {
  throw new CancelTaskException();
  }
 
@@ -833,15 +833,15 @@ private void doRun() {
  }
  }
 
- // transition into our final state. we should be either in DEPLOYING, RECOVERING,
+ // transition into our final state. we should be either in DEPLOYING, INITIALIZING,
  // RUNNING, CANCELING, or FAILED
  // loop for multiple retries during concurrent state changes via calls to cancel()
  // or to failExternally()
  while (true) {
  ExecutionState current = this.executionState;
 
  if (current == ExecutionState.RUNNING
- || current == ExecutionState.RECOVERING
+ || current == ExecutionState.INITIALIZING
  || current == ExecutionState.DEPLOYING) {
  if (t instanceof CancelTaskException) {
  if (transitionState(current, ExecutionState.CANCELED)) {
@@ -1145,7 +1145,8 @@ void cancelOrFailAndCancelInvokableInternal(ExecutionState targetState, Throwabl
  this.failureCause = cause;
  return;
  }
- } else if (current == ExecutionState.RECOVERING || current == ExecutionState.RUNNING) {
+ } else if (current == ExecutionState.INITIALIZING
+ || current == ExecutionState.RUNNING) {
  if (transitionState(current, targetState, cause)) {
  // we are canceling / failing out of the running state
  // we need to cancel the invokable
@@ -1405,7 +1406,7 @@ public void deliverOperatorEvent(OperatorID operator, SerializedValue<OperatorEv
 
  if (invokable == null
  || (executionState != ExecutionState.RUNNING
- && executionState != ExecutionState.RECOVERING)) {
+ && executionState != ExecutionState.INITIALIZING)) {
  throw new TaskNotRunningException("Task is not yet running.");
  }
 
@@ -1415,7 +1416,7 @@ public void deliverOperatorEvent(OperatorID operator, SerializedValue<OperatorEv
  ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
 
  if (getExecutionState() == ExecutionState.RUNNING
- || getExecutionState() == ExecutionState.RECOVERING) {
+ || getExecutionState() == ExecutionState.INITIALIZING) {
  FlinkException e = new FlinkException("Error while handling operator event", t);
  failExternally(e);
  throw e;

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java
@@ -93,7 +93,7 @@ public void testMixedLocalRemoteUnknownDeployment() throws Exception {
 
  // These states are allowed
  if (state == ExecutionState.RUNNING
- || state == ExecutionState.RECOVERING
+ || state == ExecutionState.INITIALIZING
  || state == ExecutionState.FINISHED
  || state == ExecutionState.SCHEDULED
  || state == ExecutionState.DEPLOYING) {