[FLINK-37401] Add JobID to MDC in JobMaster, ResourceManager, DefaultSlotStatusSyncer, FineGrainedSlotManager

Author: Savonitar

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java

@@ -470,23 +470,27 @@ protected void onStart() throws JobMasterException {
     /** Suspend the job and shutdown all other services including rpc. */
     @Override
     public CompletableFuture<Void> onStop() {
-        log.info(
-                "Stopping the JobMaster for job '{}' ({}).",
-                executionPlan.getName(),
-                executionPlan.getJobID());
-
-        // make sure there is a graceful exit
-        return stopJobExecution(
-                        new FlinkException(
-                                String.format(
-                                        "Stopping JobMaster for job '%s' (%s).",
-                                        executionPlan.getName(), executionPlan.getJobID())))
-                .exceptionally(
-                        exception -> {
-                            throw new CompletionException(
-                                    new JobMasterException(
-                                            "Could not properly stop the JobMaster.", exception));
-                        });
+        try (MdcUtils.MdcCloseable ignored =
+                MdcUtils.withContext(MdcUtils.asContextData(executionPlan.getJobID()))) {
+            log.info(
+                    "Stopping the JobMaster for job '{}' ({}).",
+                    executionPlan.getName(),
+                    executionPlan.getJobID());
+
+            // make sure there is a graceful exit
+            return stopJobExecution(
+                            new FlinkException(
+                                    String.format(
+                                            "Stopping JobMaster for job '%s' (%s).",
+                                            executionPlan.getName(), executionPlan.getJobID())))
+                    .exceptionally(
+                            exception -> {
+                                throw new CompletionException(
+                                        new JobMasterException(
+                                                "Could not properly stop the JobMaster.",
+                                                exception));
+                            });
+        }
     }
 
     // ----------------------------------------------------------------------------------------------
@@ -597,12 +601,14 @@ public void acknowledgeCheckpoint(
             final long checkpointId,
             final CheckpointMetrics checkpointMetrics,
             @Nullable final SerializedValue<TaskStateSnapshot> checkpointState) {
-        schedulerNG.acknowledgeCheckpoint(
-                jobID,
-                executionAttemptID,
-                checkpointId,
-                checkpointMetrics,
-                deserializeTaskStateSnapshot(checkpointState, getClass().getClassLoader()));
+        try (MdcUtils.MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobID))) {
+            schedulerNG.acknowledgeCheckpoint(
+                    jobID,
+                    executionAttemptID,
+                    checkpointId,
+                    checkpointMetrics,
+                    deserializeTaskStateSnapshot(checkpointState, getClass().getClassLoader()));
+        }
     }
 
     @Override
@@ -1212,9 +1218,13 @@ private CompletableFuture<Void> stopJobExecution(final Exception cause) {
         return FutureUtils.runAfterwardsAsync(
                 terminationFuture,
                 () -> {
-                    shuffleMaster.unregisterJob(executionPlan.getJobID());
-                    disconnectTaskManagerResourceManagerConnections(cause);
-                    stopJobMasterServices();
+                    try (MdcUtils.MdcCloseable ignored =
+                            MdcUtils.withContext(
+                                    MdcUtils.asContextData(executionPlan.getJobID()))) {
+                        shuffleMaster.unregisterJob(executionPlan.getJobID());
+                        disconnectTaskManagerResourceManagerConnections(cause);
+                        stopJobMasterServices();
+                    }
                 },
                 getMainThreadExecutor());
     }
@@ -1486,10 +1496,12 @@ private class ResourceManagerLeaderListener implements LeaderRetrievalListener {
         @Override
         public void notifyLeaderAddress(final String leaderAddress, final UUID leaderSessionID) {
             runAsync(
-                    () ->
-                            notifyOfNewResourceManagerLeader(
-                                    leaderAddress,
-                                    ResourceManagerId.fromUuidOrNull(leaderSessionID)));
+                    MdcUtils.wrapRunnable(
+                            MdcUtils.asContextData(executionPlan.getJobID()),
+                            () ->
+                                    notifyOfNewResourceManagerLeader(
+                                            leaderAddress,
+                                            ResourceManagerId.fromUuidOrNull(leaderSessionID))));
         }
 
         @Override
@@ -1670,13 +1682,16 @@ private class ResourceManagerHeartbeatListener implements HeartbeatListener<Void
 
         @Override
         public void notifyHeartbeatTimeout(final ResourceID resourceId) {
-            final String message =
-                    String.format(
-                            "The heartbeat of ResourceManager with id %s timed out.",
-                            resourceId.getStringWithMetadata());
-            log.info(message);
+            try (MdcUtils.MdcCloseable ignored =
+                    MdcUtils.withContext(MdcUtils.asContextData(executionPlan.getJobID()))) {
+                final String message =
+                        String.format(
+                                "The heartbeat of ResourceManager with id %s timed out.",
+                                resourceId.getStringWithMetadata());
+                log.info(message);
 
-            handleResourceManagerConnectionLoss(resourceId, new TimeoutException(message));
+                handleResourceManagerConnectionLoss(resourceId, new TimeoutException(message));
+            }
         }
 
         private void handleResourceManagerConnectionLoss(ResourceID resourceId, Exception cause) {
@@ -1691,13 +1706,16 @@ private void handleResourceManagerConnectionLoss(ResourceID resourceId, Exceptio
 
         @Override
         public void notifyTargetUnreachable(ResourceID resourceID) {
-            final String message =
-                    String.format(
-                            "ResourceManager with id %s is no longer reachable.",
-                            resourceID.getStringWithMetadata());
-            log.info(message);
+            try (MdcUtils.MdcCloseable ignored =
+                    MdcUtils.withContext(MdcUtils.asContextData(executionPlan.getJobID()))) {
+                final String message =
+                        String.format(
+                                "ResourceManager with id %s is no longer reachable.",
+                                resourceID.getStringWithMetadata());
+                log.info(message);
 
-            handleResourceManagerConnectionLoss(resourceID, new JobMasterException(message));
+                handleResourceManagerConnectionLoss(resourceID, new JobMasterException(message));
+            }
         }
 
         @Override