[flink] branch master updated: [FLINK-20139][dispatcher] Enrich logs when MiniDispatcher shutting down

[ti...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tison 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 3dc43e6  [FLINK-20139][dispatcher] Enrich logs when MiniDispatcher shutting down
3dc43e6 is described below

commit 3dc43e6fa66d1253a37e8a4bcf242b6865e340a5
Author: tison <wa...@gmail.com>
AuthorDate: Sat Nov 14 08:06:50 2020 +0800

    [FLINK-20139][dispatcher] Enrich logs when MiniDispatcher shutting down
    
    This closes #14058 .
---
 .../org/apache/flink/runtime/dispatcher/MiniDispatcher.java   | 11 ++++-------
 1 file changed, 4 insertions(+), 7 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
index b06453d..da4dd80 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
@@ -30,9 +30,6 @@ import org.apache.flink.runtime.messages.Acknowledge;
 import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.util.FlinkException;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.Collections;
 import java.util.concurrent.CompletableFuture;
 
@@ -47,7 +44,6 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
  * terminate after job completion if its execution mode is {@link ClusterEntrypoint.ExecutionMode#DETACHED}.
  */
 public class MiniDispatcher extends Dispatcher {
-	private static final Logger LOG = LoggerFactory.getLogger(MiniDispatcher.class);
 
 	private final JobClusterEntrypoint.ExecutionMode executionMode;
 	private boolean jobCancelled = false;
@@ -95,11 +91,11 @@ public class MiniDispatcher extends Dispatcher {
 				ApplicationStatus status = result.getSerializedThrowable().isPresent() ?
 						ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED;
 
-				LOG.debug("Shutting down cluster because someone retrieved the job result.");
+				log.info("Shutting down cluster because someone retrieved the job result.");
 				shutDownFuture.complete(status);
 			});
 		} else {
-			LOG.debug("Not shutting down cluster after someone retrieved the job result.");
+			log.info("Not shutting down cluster after someone retrieved the job result.");
 		}
 
 		return jobResultFuture;
@@ -117,6 +113,7 @@ public class MiniDispatcher extends Dispatcher {
 
 		if (jobCancelled || executionMode == ClusterEntrypoint.ExecutionMode.DETACHED) {
 			// shut down if job is cancelled or we don't have to wait for the execution result retrieval
+			log.info("Shutting down cluster with state {}, jobCancelled: {}, executionMode: {}", archivedExecutionGraph.getState(), jobCancelled, executionMode);
 			shutDownFuture.complete(ApplicationStatus.fromJobStatus(archivedExecutionGraph.getState()));
 		}
 	}
@@ -124,8 +121,8 @@ public class MiniDispatcher extends Dispatcher {
 	@Override
 	protected void jobNotFinished(JobID jobId) {
 		super.jobNotFinished(jobId);
-
 		// shut down since we have done our job
+		log.info("Shutting down cluster because job not finished");
 		shutDownFuture.complete(ApplicationStatus.UNKNOWN);
 	}
 }