[flink] branch release-1.12 updated: [FLINK-20351] Correct the logging of the location of a task failure on the JM

[trohrmann]

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

trohrmann pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.12 by this push:
     new 84b6631  [FLINK-20351] Correct the logging of the location of a task 
failure on the JM
84b6631 is described below

commit 84b6631eee9207c5289df9e185a6c19dd84ec9fd
Author: Till Rohrmann <trohrm...@apache.org>
AuthorDate: Wed Nov 25 18:21:40 2020 +0100

    [FLINK-20351] Correct the logging of the location of a task failure on the 
JM
    
    This commit logs the TaskManagerLocation.toString instead of the 
LogicalSlot.toString
    when logging an Execution failure.
    
    This closes #14225.
---
 .../org/apache/flink/runtime/executiongraph/Execution.java   | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

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 51fda87..fb2bd48 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
@@ -1606,15 +1606,13 @@ public class Execution implements AccessExecution, 
Archiveable<ArchivedExecution
                                LOG.info("{} ({}) switched from {} to {}.", 
getVertex().getTaskNameWithSubtaskIndex(), getAttemptId(), currentState, 
targetState);
                        } else {
                                if (LOG.isInfoEnabled()) {
-                                       final String locationInformation = 
getAssignedResource() != null ? getAssignedResource().toString() : "not 
deployed";
-
                                        LOG.info(
                                                "{} ({}) switched from {} to {} 
on {}.",
                                                
getVertex().getTaskNameWithSubtaskIndex(),
                                                getAttemptId(),
                                                currentState,
                                                targetState,
-                                               locationInformation,
+                                               getLocationInformation(),
                                                error);
                                }
                        }
@@ -1638,6 +1636,14 @@ public class Execution implements AccessExecution, 
Archiveable<ArchivedExecution
                }
        }
 
+       private String getLocationInformation() {
+               if (assignedResource != null) {
+                       return 
assignedResource.getTaskManagerLocation().toString();
+               } else {
+                       return "[unassigned resource]";
+               }
+       }
+
        private void markTimestamp(ExecutionState state) {
                markTimestamp(state, System.currentTimeMillis());
        }