[FLINK-17300] Log the lineage information between ExecutionAttemptID …

…and AllocationID This closes apache#11852.
spotify · Jun 25, 2020 · 7e48549 · 7e48549
1 parent 581fabe
commit 7e48549
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 6 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
@@ -728,10 +728,8 @@ public void deploy() throws JobException {
  return;
  }
 
- if (LOG.isInfoEnabled()) {
- LOG.info(String.format("Deploying %s (attempt #%d) to %s", vertex.getTaskNameWithSubtaskIndex(),
- attemptNumber, getAssignedResourceLocation()));
- }
+ LOG.info("Deploying {} (attempt #{}) with attempt id {} to {} with allocation id {}", vertex.getTaskNameWithSubtaskIndex(),
+ attemptNumber, vertex.getCurrentExecutionAttempt().getAttemptId(), getAssignedResourceLocation(), slot.getAllocationId());
 
  final TaskDeploymentDescriptor deployment = TaskDeploymentDescriptorFactory
  .fromExecutionVertex(vertex, attemptNumber)

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
@@ -643,7 +643,8 @@ public CompletableFuture<Acknowledge> submitTask(
 
  taskMetricGroup.gauge(MetricNames.IS_BACKPRESSURED, task::isBackPressured);
 
- log.info("Received task {}.", task.getTaskInfo().getTaskNameWithSubtasks());
+ log.info("Received task {} ({}), deploy into slot with allocation id {}.",
+ task.getTaskInfo().getTaskNameWithSubtasks(), tdd.getExecutionAttemptId(), tdd.getAllocationId());
 
  boolean taskAdded;
 

diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -573,7 +573,7 @@ public boolean accept(File dir, String name) {
  String expected = "Starting TaskManagers";
  Assert.assertTrue("Expected string '" + expected + "' not found in JobManager log: '" + jobmanagerLog + "'",
  content.contains(expected));
- expected = " (2/2) (attempt #0) to ";
+ expected = " (2/2) (attempt #0) with attempt id ";
  Assert.assertTrue("Expected string '" + expected + "' not found in JobManager log." +
  "This string checks that the job has been started with a parallelism of 2. Log contents: '" + jobmanagerLog + "'",
  content.contains(expected));