[FLINK-40208][runtime] Add test coverage and blank-key validation for JobMdcRegistry

docs/content.zh/docs/ops/logging_context.md

@@ -1,6 +1,6 @@
 ---
 title: "Logging Context (MDC)"
-weight: 7
+weight: 9
 type: docs
 ---
 <!--

docs/content/docs/ops/logging_context.md

@@ -1,6 +1,6 @@
 ---
 title: "Logging Context (MDC)"
-weight: 7
+weight: 9
 type: docs
 ---
 <!--

flink-core/src/main/java/org/apache/flink/util/MdcUtils.java

@@ -18,27 +18,44 @@
 
 package org.apache.flink.util;
 
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.MdcOptions;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
+import javax.annotation.concurrent.ThreadSafe;
+
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
 
 /** Utility class to manage common Flink attributes in {@link MDC}. */
+@ThreadSafe
 public class MdcUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(MdcUtils.class);
 
     public static final String JOB_ID = "flink-job-id";
 
+    private static final Set<String> WARNED_BLANK_MDC_KEYS = ConcurrentHashMap.newKeySet();
+
+    /** Resets the deduplication set for blank MDC key name warnings. */
+    @VisibleForTesting
+    static void clearWarnedKeyNames() {
+        WARNED_BLANK_MDC_KEYS.clear();
+    }
+
     /**
      * Replace MDC contents with the provided one and return a closeable object that can be used to
      * restore the original MDC.
@@ -138,9 +155,18 @@ public static Map<String, String> asContextData(
                 jobConfiguration.get(MdcOptions.JOB_CONFIGURATION_TO_MDC_KEYS);
         final Map<String, String> context = new HashMap<>();
         for (Map.Entry<String, String> entry : mdcKeyMapping.entrySet()) {
+            final String mdcKeyName = entry.getValue();
+            if (mdcKeyName == null || mdcKeyName.isBlank()) {
+                if (WARNED_BLANK_MDC_KEYS.add(entry.getKey())) {
+                    LOG.warn(
+                            "MDC key name for configuration key '{}' is blank; skipping.",
+                            entry.getKey());
+                }
+                continue;
+            }
             final String value = jobConfiguration.getString(entry.getKey(), null);
             if (value != null && !value.isBlank()) {
-                context.put(entry.getValue(), value);
+                context.put(mdcKeyName, value);
             }
         }
         if (context.isEmpty()) {

flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java

@@ -41,6 +41,7 @@
 import org.slf4j.MDC;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
@@ -66,11 +67,16 @@ class MdcUtilsTest {
     public final LoggerAuditingExtension loggerExtension =
             new LoggerAuditingExtension(MdcUtilsTest.class, DEBUG);
 
+    @RegisterExtension
+    final LoggerAuditingExtension mdcUtilsLogExtension =
+            new LoggerAuditingExtension(MdcUtils.class, org.slf4j.event.Level.WARN);
+
     @BeforeEach
     @AfterEach
     void clearMdcAndRegistry() {
         MDC.clear();
         JobMdcRegistry.clear();
+        MdcUtils.clearWarnedKeyNames();
     }
 
     @Test
@@ -244,6 +250,39 @@ void testKeySkippedWhenValueAbsentOrBlank(final String scenario, final String co
                 .isEqualTo(Collections.singletonMap(MdcUtils.JOB_ID, jobID.toHexString()));
     }
 
+    private static Stream<Arguments> blankMdcKeyNameCases() {
+        return Stream.of(
+                Arguments.of("empty string", ""),
+                Arguments.of("whitespace only", "   "),
+                Arguments.of("null key name", (String) null));
+    }
+
+    @ParameterizedTest
+    @MethodSource("blankMdcKeyNameCases")
+    void testBlankMdcKeyNameSkippedWithWarning(final String scenario, final String mdcKeyName) {
+        final JobID jobID = new JobID();
+        final Configuration conf = new Configuration();
+        final Map<String, String> keyMapping = new HashMap<>();
+        keyMapping.put("job.key-1", mdcKeyName);
+        conf.set(MdcOptions.JOB_CONFIGURATION_TO_MDC_KEYS, keyMapping);
+        conf.setString("job.key-1", "val-1");
+
+        // First call — should log WARN and skip the blank key
+        final Map<String, String> context1 = MdcUtils.asContextData(jobID, conf);
+        assertThat(context1)
+                .as(scenario + " — first call")
+                .isEqualTo(Collections.singletonMap(MdcUtils.JOB_ID, jobID.toHexString()));
+
+        // Second call — same blank key, WARN must NOT repeat
+        final Map<String, String> context2 = MdcUtils.asContextData(jobID, conf);
+        assertThat(context2)
+                .as(scenario + " — second call")
+                .isEqualTo(Collections.singletonMap(MdcUtils.JOB_ID, jobID.toHexString()));
+
+        // Assert WARN logged exactly once
+        assertThat(mdcUtilsLogExtension.getEvents()).as(scenario + " — WARN count").hasSize(1);
+    }
+
     // --- JobMdcRegistry integration: registry-first lookup ---
 
     @Test

flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java

@@ -107,6 +107,7 @@
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.JobMdcRegistry;
+import org.apache.flink.util.MdcUtils;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.concurrent.FutureUtils;
 
@@ -572,12 +573,7 @@ public void testCancellationOfNonCanceledTerminalJobFailsWithAppropriateExceptio
 
     @Test
     public void testJobMdcContextRegisteredOnSubmissionAndClearedOnTermination() throws Exception {
-        final Map<String, String> keyMapping = new HashMap<>();
-        keyMapping.put("job.key-1", "mdc-key-1");
-        keyMapping.put("job.key-2", "mdc-key-2");
-        jobGraph.getJobConfiguration().set(MdcOptions.JOB_CONFIGURATION_TO_MDC_KEYS, keyMapping);
-        jobGraph.getJobConfiguration().setString("job.key-1", "val-1");
-        jobGraph.getJobConfiguration().setString("job.key-2", "val-2");
+        configureJobWithMdcEnrichment();
 
         final CompletableFuture<JobManagerRunnerResult> resultFuture = new CompletableFuture<>();
         dispatcher =
@@ -611,6 +607,32 @@ public void testJobMdcContextRegisteredOnSubmissionAndClearedOnTermination() thr
         CommonTestUtils.waitUntilCondition(() -> JobMdcRegistry.lookup(jobId) == null);
     }
 
+    @Test
+    public void testJobMdcContextRegisteredOnRecovery() throws Exception {
+        configureJobWithMdcEnrichment();
+
+        jobMasterLeaderElection.isLeader(UUID.randomUUID());
+
+        final TestingJobMasterServiceLeadershipRunnerFactory runnerFactory =
+                new TestingJobMasterServiceLeadershipRunnerFactory();
+        dispatcher =
+                createTestingDispatcherBuilder()
+                        .setJobManagerRunnerFactory(runnerFactory)
+                        .setRecoveredJobs(Collections.singleton(jobGraph))
+                        .build(rpcService);
+        dispatcher.start();
+
+        // takeCreatedJobManagerRunner blocks until the runner is created,
+        // which happens AFTER registerOrClear in runRecoveredJob
+        runnerFactory.takeCreatedJobManagerRunner();
+
+        assertThat(JobMdcRegistry.lookup(jobId))
+                .containsEntry(MdcUtils.JOB_ID, jobId.toHexString())
+                .containsEntry("mdc-key-1", "val-1")
+                .containsEntry("mdc-key-2", "val-2")
+                .hasSize(3);
+    }
+
     @Test
     public void testNoHistoryServerArchiveCreatedForSuspendedJob() throws Exception {
         final CompletableFuture<Void> archiveAttemptFuture = new CompletableFuture<>();
@@ -675,6 +697,15 @@ private void mockApplicationFinished() throws Exception {
                 .get();
     }
 
+    private void configureJobWithMdcEnrichment() {
+        final Map<String, String> keyMapping = new HashMap<>();
+        keyMapping.put("job.key-1", "mdc-key-1");
+        keyMapping.put("job.key-2", "mdc-key-2");
+        jobGraph.getJobConfiguration().set(MdcOptions.JOB_CONFIGURATION_TO_MDC_KEYS, keyMapping);
+        jobGraph.getJobConfiguration().setString("job.key-1", "val-1");
+        jobGraph.getJobConfiguration().setString("job.key-2", "val-2");
+    }
+
     @Test
     public void testJobManagerRunnerInitializationFailureFailsJob() throws Exception {
         final TestingJobMasterServiceLeadershipRunnerFactory testingJobManagerRunnerFactory =