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 2020/11/09 09:11:09 UTC
[flink] 02/03: [FLINK-20033] Make JobManagerJobStatusListener call
back run directly
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 90fb07c311aa6c91ad544705ee96a14ab9302122
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Sat Nov 7 17:20:25 2020 +0100
[FLINK-20033] Make JobManagerJobStatusListener call back run directly
Since the ExecutionGraph can only be modified using the JobMaster's main thread, we no longer
need to trigger a jobStatusChanged async message when the JobManagerJobStatusListener.jobStatusChanges
callback is triggered. This ensures that job status changes will directly be reported.
---
.../org/apache/flink/runtime/jobmaster/JobMaster.java | 16 +---------------
1 file changed, 1 insertion(+), 15 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 883b313..3f0c701 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -921,10 +921,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
if (jobManagerJobMetricGroup != null) {
jobManagerJobMetricGroup.close();
}
-
- if (jobStatusListener != null) {
- jobStatusListener.stop();
- }
}
private void clearSchedulerFields() {
@@ -1178,23 +1174,13 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private class JobManagerJobStatusListener implements JobStatusListener {
- private volatile boolean running = true;
-
@Override
public void jobStatusChanges(
final JobID jobId,
final JobStatus newJobStatus,
final long timestamp,
final Throwable error) {
-
- if (running) {
- // run in rpc thread to avoid concurrency
- runAsync(() -> jobStatusChanged(newJobStatus, timestamp, error));
- }
- }
-
- private void stop() {
- running = false;
+ jobStatusChanged(newJobStatus, timestamp, error);
}
}