You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2021/04/13 14:18:25 UTC
[flink] branch master updated: [FLINK-22215][runtime] RECOVERING
state is renamed to INITIALIZING
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b387928 [FLINK-22215][runtime] RECOVERING state is renamed to INITIALIZING
b387928 is described below
commit b3879280fa60f11c874af042b81ff5ac4672a3eb
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Mon Apr 12 13:05:02 2021 +0200
[FLINK-22215][runtime] RECOVERING state is renamed to INITIALIZING
---
.../shortcodes/generated/rest_v1_dispatcher.html | 10 ++++-----
.../src/test/resources/rest_api_v1.snapshot | 10 ++++-----
.../web-dashboard/src/app/app.config.ts | 3 +--
.../web-dashboard/src/app/interfaces/job-detail.ts | 2 +-
.../src/app/interfaces/job-overview.ts | 2 +-
.../src/app/interfaces/job-vertex-task-manager.ts | 2 +-
.../TaskDeploymentDescriptorFactory.java | 2 +-
.../flink/runtime/execution/ExecutionState.java | 6 +++---
.../executiongraph/DefaultExecutionGraph.java | 4 ++--
.../flink/runtime/executiongraph/Execution.java | 20 +++++++++---------
.../runtime/executiongraph/ExecutionJobVertex.java | 4 ++--
.../consumer/RemoteChannelStateChecker.java | 4 ++--
.../DefaultOperatorCoordinatorHandler.java | 2 +-
.../org/apache/flink/runtime/taskmanager/Task.java | 21 ++++++++++---------
.../runtime/deployment/ShuffleDescriptorTest.java | 2 +-
.../executiongraph/ArchivedExecutionGraphTest.java | 8 ++++----
.../scheduler/UpdatePartitionConsumersTest.java | 8 ++++----
.../flink/runtime/jobmaster/JobMasterTest.java | 2 +-
.../OperatorCoordinatorSchedulerTest.java | 2 +-
.../DefaultSchedulerBatchSchedulingTest.java | 2 +-
.../runtime/scheduler/DefaultSchedulerTest.java | 2 +-
.../runtime/scheduler/SchedulerTestingUtils.java | 2 +-
.../apache/flink/runtime/taskmanager/TaskTest.java | 24 +++++++++++-----------
23 files changed, 72 insertions(+), 72 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index 75b8cce..31da72e 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -1558,7 +1558,7 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"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 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"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 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"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 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"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 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"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
index 7589369..bf1f190 100644
--- 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
index e4e2c7b..97a5c83 100644
--- 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
index 4483a41..040039a 100644
--- 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
index 05e93d6..6f7d64e 100644
--- 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
index 6f3e104..3877712 100644
--- 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/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java
index f467fb0..c0b543a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java
@@ -258,7 +258,7 @@ public class TaskDeploymentDescriptorFactory {
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
index 3c0d4e9..e25d639 100644
--- 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 @@ package org.apache.flink.runtime.execution;
* 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 @@ package org.apache.flink.runtime.execution;
* |
* | ... -> 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/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
index fdb18ce..26ac38f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -957,7 +957,7 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
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 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
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
index d45ce8a..a39000a 100644
--- 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.CREATED;
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 class Execution
}
// 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 @@ public class Execution
public CompletableFuture<?> suspend() {
switch (state) {
case RUNNING:
- case RECOVERING:
+ case INITIALIZING:
case DEPLOYING:
case CREATED:
case SCHEDULED:
@@ -723,7 +723,7 @@ public class Execution
// ----------------------------------------------------------------
if (consumerState == DEPLOYING
|| consumerState == RUNNING
- || consumerState == RECOVERING) {
+ || consumerState == INITIALIZING) {
final PartitionInfo partitionInfo = createPartitionInfo(partition);
if (consumerState == DEPLOYING) {
@@ -856,7 +856,7 @@ public class Execution
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 @@ public class Execution
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 @@ public class Execution
return;
} else if (current == CANCELING
|| current == RUNNING
- || current == RECOVERING
+ || current == INITIALIZING
|| current == DEPLOYING) {
updateAccumulatorsAndMetrics(userAccumulators, metrics);
@@ -1138,7 +1138,7 @@ public class Execution
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 @@ public class Execution
}
boolean switchToRecovering() {
- if (switchTo(DEPLOYING, RECOVERING)) {
+ if (switchTo(DEPLOYING, INITIALIZING)) {
sendPartitionInfos();
return true;
}
@@ -1168,7 +1168,7 @@ public class Execution
}
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
index b3d065a..04e3a43 100644
--- 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 class ExecutionJobVertex
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/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteChannelStateChecker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteChannelStateChecker.java
index f32ee84..899d2b9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteChannelStateChecker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteChannelStateChecker.java
@@ -67,7 +67,7 @@ public class RemoteChannelStateChecker {
private static boolean isConsumerStateValidForConsumption(
ExecutionState consumerExecutionState) {
return consumerExecutionState == ExecutionState.RUNNING
- || consumerExecutionState == ExecutionState.RECOVERING
+ || consumerExecutionState == ExecutionState.INITIALIZING
|| consumerExecutionState == ExecutionState.DEPLOYING;
}
@@ -75,7 +75,7 @@ public class RemoteChannelStateChecker {
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/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
index e660fd5..245e92e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
@@ -112,7 +112,7 @@ public class DefaultOperatorCoordinatorHandler implements OperatorCoordinatorHan
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
index 0f5bd95..acc0b76 100644
--- 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 @@ public class Task
public boolean isBackPressured() {
if (invokable == null
|| consumableNotifyingPartitionWriters.length == 0
- || (executionState != ExecutionState.RECOVERING
+ || (executionState != ExecutionState.INITIALIZING
&& executionState != ExecutionState.RUNNING)) {
return false;
}
@@ -738,14 +738,14 @@ public class Task
// 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 @@ public class Task
FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
}
- if (!transitionState(ExecutionState.RECOVERING, ExecutionState.RUNNING)) {
+ if (!transitionState(ExecutionState.INITIALIZING, ExecutionState.RUNNING)) {
throw new CancelTaskException();
}
@@ -833,7 +833,7 @@ public class Task
}
}
- // 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()
@@ -841,7 +841,7 @@ public class Task
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 @@ public class Task
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 class Task
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 class Task
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
index 827be77..5fc142c 100644
--- 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 class ShuffleDescriptorTest extends TestLogger {
// These states are allowed
if (state == ExecutionState.RUNNING
- || state == ExecutionState.RECOVERING
+ || state == ExecutionState.INITIALIZING
|| state == ExecutionState.FINISHED
|| state == ExecutionState.SCHEDULED
|| state == ExecutionState.DEPLOYING) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
index 65fddee..cace7ee 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
@@ -350,8 +350,8 @@ public class ArchivedExecutionGraphTest extends TestLogger {
runtimeVertex.getStateTimestamp(ExecutionState.DEPLOYING),
archivedVertex.getStateTimestamp(ExecutionState.DEPLOYING));
assertEquals(
- runtimeVertex.getStateTimestamp(ExecutionState.RECOVERING),
- archivedVertex.getStateTimestamp(ExecutionState.RECOVERING));
+ runtimeVertex.getStateTimestamp(ExecutionState.INITIALIZING),
+ archivedVertex.getStateTimestamp(ExecutionState.INITIALIZING));
assertEquals(
runtimeVertex.getStateTimestamp(ExecutionState.RUNNING),
archivedVertex.getStateTimestamp(ExecutionState.RUNNING));
@@ -408,8 +408,8 @@ public class ArchivedExecutionGraphTest extends TestLogger {
runtimeExecution.getStateTimestamp(ExecutionState.DEPLOYING),
archivedExecution.getStateTimestamp(ExecutionState.DEPLOYING));
assertEquals(
- runtimeExecution.getStateTimestamp(ExecutionState.RECOVERING),
- archivedExecution.getStateTimestamp(ExecutionState.RECOVERING));
+ runtimeExecution.getStateTimestamp(ExecutionState.INITIALIZING),
+ archivedExecution.getStateTimestamp(ExecutionState.INITIALIZING));
assertEquals(
runtimeExecution.getStateTimestamp(ExecutionState.RUNNING),
archivedExecution.getStateTimestamp(ExecutionState.RUNNING));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/UpdatePartitionConsumersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/UpdatePartitionConsumersTest.java
index 3217d42..ee58347 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/UpdatePartitionConsumersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/UpdatePartitionConsumersTest.java
@@ -155,13 +155,13 @@ public class UpdatePartitionConsumersTest extends TestLogger {
assertThat(ev3.getExecutionState(), is(ExecutionState.DEPLOYING));
assertThat(ev4.getExecutionState(), is(ExecutionState.DEPLOYING));
- updateState(scheduler, ev1, ExecutionState.RECOVERING);
+ updateState(scheduler, ev1, ExecutionState.INITIALIZING);
updateState(scheduler, ev1, ExecutionState.RUNNING);
- updateState(scheduler, ev2, ExecutionState.RECOVERING);
+ updateState(scheduler, ev2, ExecutionState.INITIALIZING);
updateState(scheduler, ev2, ExecutionState.RUNNING);
- updateState(scheduler, ev3, ExecutionState.RECOVERING);
+ updateState(scheduler, ev3, ExecutionState.INITIALIZING);
updateState(scheduler, ev3, ExecutionState.RUNNING);
- updateState(scheduler, ev4, ExecutionState.RECOVERING);
+ updateState(scheduler, ev4, ExecutionState.INITIALIZING);
updateState(scheduler, ev4, ExecutionState.RUNNING);
final InputGateDeploymentDescriptor ev4Igdd2 =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index 14ccab5..f274e2f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -1728,7 +1728,7 @@ public class JobMasterTest extends TestLogger {
jobMasterGateway
.updateTaskExecutionState(
- new TaskExecutionState(executionAttemptId, ExecutionState.RECOVERING))
+ new TaskExecutionState(executionAttemptId, ExecutionState.INITIALIZING))
.get();
jobMasterGateway
.updateTaskExecutionState(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java
index ebc598a..b256185 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java
@@ -751,7 +751,7 @@ public class OperatorCoordinatorSchedulerTest extends TestLogger {
private void failAndRestartTask(DefaultScheduler scheduler, int subtask) {
failAndRedeployTask(scheduler, subtask);
- setExecutionToState(ExecutionState.RECOVERING, scheduler, testVertexId, subtask);
+ setExecutionToState(ExecutionState.INITIALIZING, scheduler, testVertexId, subtask);
setExecutionToState(ExecutionState.RUNNING, scheduler, testVertexId, subtask);
// guard the test assumptions: This must bring the task back to RUNNING
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBatchSchedulingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBatchSchedulingTest.java
index b5bb87a..56dfc6e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBatchSchedulingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBatchSchedulingTest.java
@@ -164,7 +164,7 @@ public class DefaultSchedulerBatchSchedulingTest extends TestLogger {
() -> {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
- executionAttemptId, ExecutionState.RECOVERING));
+ executionAttemptId, ExecutionState.INITIALIZING));
scheduler.updateTaskExecutionState(
new TaskExecutionState(
executionAttemptId, ExecutionState.RUNNING));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
index f45ca0c..31f5a42 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
@@ -1327,7 +1327,7 @@ public class DefaultSchedulerTest extends TestLogger {
private void transitionToRunning(DefaultScheduler scheduler, ExecutionAttemptID attemptId) {
Preconditions.checkState(
scheduler.updateTaskExecutionState(
- new TaskExecutionState(attemptId, ExecutionState.RECOVERING)));
+ new TaskExecutionState(attemptId, ExecutionState.INITIALIZING)));
Preconditions.checkState(
scheduler.updateTaskExecutionState(
new TaskExecutionState(attemptId, ExecutionState.RUNNING)));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
index 1027e01..ac5b1cd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
@@ -235,7 +235,7 @@ public class SchedulerTestingUtils {
.forEach(
(attemptId) -> {
scheduler.updateTaskExecutionState(
- new TaskExecutionState(attemptId, ExecutionState.RECOVERING));
+ new TaskExecutionState(attemptId, ExecutionState.INITIALIZING));
scheduler.updateTaskExecutionState(
new TaskExecutionState(attemptId, ExecutionState.RUNNING));
});
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index 1df77eb..d812508 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -137,7 +137,7 @@ public class TaskTest extends TestLogger {
assertNull(task.getFailureCause());
assertNull(task.getInvokable());
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
@@ -307,7 +307,7 @@ public class TaskTest extends TestLogger {
assertNotNull(task.getFailureCause().getMessage());
assertThat(task.getFailureCause().getMessage(), containsString(RESTORE_EXCEPTION_MSG));
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new Exception(RESTORE_EXCEPTION_MSG));
}
@@ -329,7 +329,7 @@ public class TaskTest extends TestLogger {
assertNotNull(task.getFailureCause().getMessage());
assertTrue(task.getFailureCause().getMessage().contains("test"));
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new Exception("test"));
@@ -352,7 +352,7 @@ public class TaskTest extends TestLogger {
final Throwable cause = task.getFailureCause();
assertTrue(cause instanceof IOException);
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new IOException("test"));
@@ -384,7 +384,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertNull(task.getFailureCause());
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.CANCELED, task, null);
}
@@ -414,7 +414,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertNull(task.getFailureCause());
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.CANCELED, task, null);
}
@@ -442,7 +442,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertThat(task.getFailureCause().getMessage(), containsString(RESTORE_EXCEPTION_MSG));
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new Exception(RESTORE_EXCEPTION_MSG));
}
@@ -470,7 +470,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertTrue(task.getFailureCause().getMessage().contains("test"));
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new Exception("test"));
@@ -494,7 +494,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertTrue(task.getFailureCause().getMessage().contains("test"));
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new Exception("test"));
@@ -528,7 +528,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertNull(task.getFailureCause());
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.CANCELED, task, null);
}
@@ -560,7 +560,7 @@ public class TaskTest extends TestLogger {
assertTrue(task.isCanceledOrFailed());
assertTrue(task.getFailureCause().getMessage().contains("external"));
- taskManagerActions.validateListenerMessage(ExecutionState.RECOVERING, task, null);
+ taskManagerActions.validateListenerMessage(ExecutionState.INITIALIZING, task, null);
taskManagerActions.validateListenerMessage(ExecutionState.RUNNING, task, null);
taskManagerActions.validateListenerMessage(
ExecutionState.FAILED, task, new Exception("external"));
@@ -638,7 +638,7 @@ public class TaskTest extends TestLogger {
expected.put(state, ExecutionState.FAILED);
}
- expected.put(ExecutionState.RECOVERING, initialTaskState);
+ expected.put(ExecutionState.INITIALIZING, initialTaskState);
expected.put(ExecutionState.RUNNING, initialTaskState);
expected.put(ExecutionState.SCHEDULED, initialTaskState);
expected.put(ExecutionState.DEPLOYING, initialTaskState);