You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/10/09 08:24:43 UTC
[flink] 04/05: [FLINK-10513] Replace
TaskManagerActions#notifyFinalState with #updateTaskExecutionState
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f3bbbb8c218eb3eada239b281092e4bd97d4ebb3
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 8 12:24:34 2018 +0200
[FLINK-10513] Replace TaskManagerActions#notifyFinalState with #updateTaskExecutionState
Simplify TaskManagerActions interface by replacing calls to notifyFinalState with
updateTaskExecutionState.
This closes #6804.
---
.../apache/flink/runtime/taskexecutor/TaskExecutor.java | 11 +++++------
.../taskmanager/ActorGatewayTaskManagerActions.java | 16 ++++++++--------
.../java/org/apache/flink/runtime/taskmanager/Task.java | 2 +-
.../flink/runtime/taskmanager/TaskManagerActions.java | 7 -------
.../runtime/taskmanager/NoOpTaskManagerActions.java | 3 ---
.../org/apache/flink/runtime/taskmanager/TaskTest.java | 6 ------
6 files changed, 14 insertions(+), 31 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index e641749..e2ec482 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -1553,11 +1553,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
@Override
- public void notifyFinalState(TaskExecutionState taskExecutionState) {
- runAsync(() -> unregisterTaskAndNotifyFinalState(jobMasterGateway, taskExecutionState.getID()));
- }
-
- @Override
public void notifyFatalError(String message, Throwable cause) {
try {
log.error(message, cause);
@@ -1574,7 +1569,11 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
@Override
public void updateTaskExecutionState(final TaskExecutionState taskExecutionState) {
- TaskExecutor.this.updateTaskExecutionState(jobMasterGateway, taskExecutionState);
+ if (taskExecutionState.getExecutionState().isTerminal()) {
+ runAsync(() -> unregisterTaskAndNotifyFinalState(jobMasterGateway, taskExecutionState.getID()));
+ } else {
+ TaskExecutor.this.updateTaskExecutionState(jobMasterGateway, taskExecutionState);
+ }
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ActorGatewayTaskManagerActions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ActorGatewayTaskManagerActions.java
index bb1ba86..1731fcc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ActorGatewayTaskManagerActions.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ActorGatewayTaskManagerActions.java
@@ -36,11 +36,6 @@ public class ActorGatewayTaskManagerActions implements TaskManagerActions {
}
@Override
- public void notifyFinalState(TaskExecutionState taskExecutionState) {
- actorGateway.tell(new TaskMessages.TaskInFinalState(taskExecutionState.getID()));
- }
-
- @Override
public void notifyFatalError(String message, Throwable cause) {
actorGateway.tell(new TaskManagerMessages.FatalError(message, cause));
}
@@ -52,8 +47,13 @@ public class ActorGatewayTaskManagerActions implements TaskManagerActions {
@Override
public void updateTaskExecutionState(TaskExecutionState taskExecutionState) {
- TaskMessages.UpdateTaskExecutionState actorMessage = new TaskMessages.UpdateTaskExecutionState(taskExecutionState);
-
- actorGateway.tell(actorMessage);
+ final TaskMessages.TaskMessage taskMessage;
+ if (taskExecutionState.getExecutionState().isTerminal()) {
+ taskMessage = new TaskMessages.TaskInFinalState(taskExecutionState.getID());
+ } else {
+ taskMessage = new TaskMessages.UpdateTaskExecutionState(taskExecutionState);
+ }
+
+ actorGateway.tell(taskMessage);
}
}
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 d8c5834..9abf92f 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
@@ -869,7 +869,7 @@ public class Task implements Runnable, TaskActions, CheckpointListener {
private void notifyFinalState() {
checkState(executionState.isTerminal());
- taskManagerActions.notifyFinalState(new TaskExecutionState(jobId, executionId, executionState, failureCause));
+ taskManagerActions.updateTaskExecutionState(new TaskExecutionState(jobId, executionId, executionState, failureCause));
}
private void notifyFatalError(String message, Throwable cause) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerActions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerActions.java
index b1c4017..09242c2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerActions.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerActions.java
@@ -26,13 +26,6 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
public interface TaskManagerActions {
/**
- * Notifies the task manager that the given task is in a final state.
- *
- * @param taskExecutionState final task execution state
- */
- void notifyFinalState(TaskExecutionState taskExecutionState);
-
- /**
* Notifies the task manager about a fatal error occurred in the task.
*
* @param message Message to report
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpTaskManagerActions.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpTaskManagerActions.java
index b9fb202..ea302c0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpTaskManagerActions.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpTaskManagerActions.java
@@ -26,9 +26,6 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
public class NoOpTaskManagerActions implements TaskManagerActions {
@Override
- public void notifyFinalState(TaskExecutionState taskExecutionState) {}
-
- @Override
public void notifyFatalError(String message, Throwable cause) {}
@Override
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 e8f7f58..bb5cd17 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
@@ -134,12 +134,6 @@ public class TaskTest extends TestLogger {
super.updateTaskExecutionState(taskExecutionState);
listenerMessages.add(taskExecutionState);
}
-
- @Override
- public void notifyFinalState(TaskExecutionState taskExecutionState) {
- super.notifyFinalState(taskExecutionState);
- listenerMessages.add(taskExecutionState);
- }
};
awaitLatch = new OneShotLatch();