[hotfix] Add debug logging to the states of the AdaptiveScheduler

This commit adds debug log statements to the states of the AdaptiveScheduler to log
whenever we ignore a global failure.

Author: tillrohrmann

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java

@@ -1034,14 +1034,24 @@ private static int getCumulativeParallelism(VertexParallelism potentialNewParall
 
     @Override
     public void onFinished(ArchivedExecutionGraph archivedExecutionGraph) {
+
+        @Nullable
+        final Throwable optionalFailure =
+                archivedExecutionGraph.getFailureInfo() != null
+                        ? archivedExecutionGraph.getFailureInfo().getException()
+                        : null;
+        LOG.info(
+                "Job {} reached terminal state {}.",
+                archivedExecutionGraph.getJobID(),
+                archivedExecutionGraph.getState(),
+                optionalFailure);
+
         if (jobStatusListener != null) {
             jobStatusListener.jobStatusChanges(
                     jobInformation.getJobID(),
                     archivedExecutionGraph.getState(),
                     archivedExecutionGraph.getStatusTimestamp(archivedExecutionGraph.getState()),
-                    archivedExecutionGraph.getFailureInfo() != null
-                            ? archivedExecutionGraph.getFailureInfo().getException()
-                            : null);
+                    optionalFailure);
         }
 
         jobTerminationFuture.complete(archivedExecutionGraph.getState());

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Canceling.java

@@ -56,7 +56,8 @@ public void cancel() {
 
     @Override
     public void handleGlobalFailure(Throwable cause) {
-        // ignore global failures
+        getLogger()
+                .debug("Ignored global failure because we are already canceling the job.", cause);
     }
 
     @Override

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Failing.java

@@ -58,7 +58,7 @@ public void cancel() {
 
     @Override
     public void handleGlobalFailure(Throwable cause) {
-        // nothing to do since we are already failing
+        getLogger().debug("Ignored global failure because we are already failing the job.", cause);
     }
 
     @Override

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java

@@ -54,7 +54,12 @@ public ArchivedExecutionGraph getJob() {
     }
 
     @Override
-    public void handleGlobalFailure(Throwable cause) {}
+    public void handleGlobalFailure(Throwable cause) {
+        logger.debug(
+                "Ignore global failure because we are in state {}.",
+                getClass().getSimpleName(),
+                cause);
+    }
 
     @Override
     public Logger getLogger() {

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java

@@ -77,7 +77,8 @@ public void cancel() {
 
     @Override
     public void handleGlobalFailure(Throwable cause) {
-        // don't do anything
+        getLogger()
+                .debug("Ignored global failure because we are already restarting the job.", cause);
     }
 
     @Override