Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Izeren wants to merge 1 commit into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/content.zh/docs/ops/logging_context.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
---
title: "Logging Context (MDC)"
weight: 7
weight: 9
type: docs
---
<!--
Expand Down
2 changes: 1 addition & 1 deletion docs/content/docs/ops/logging_context.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
---
title: "Logging Context (MDC)"
weight: 7
weight: 9
type: docs
---
<!--
Expand Down
27 changes: 26 additions & 1 deletion flink-core/src/main/java/org/apache/flink/util/MdcUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,30 +18,46 @@

package org.apache.flink.util;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobInfo;
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.Nonnull;
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();
}

/**
* Longest job name embedded in a thread name; longer ones, such as generated SQL job names, are
* truncated. Matches the length of the hex {@link JobID} that follows it.
Expand Down Expand Up @@ -162,9 +178,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;
Comment on lines +182 to +188

@rkhachatryan rkhachatryan Aug 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

TBH this solutions seems a bit too complex for the problem to me..
What if we instead interpret empty mdcKeyName as an intention to use entry.getKey() for both job config parameter and logging key?

That adds some convenience and eliminates the problem with empty keys.

WDYT?

}
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()) {
Expand Down
39 changes: 39 additions & 0 deletions flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,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;
Expand All @@ -68,11 +69,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
Expand Down Expand Up @@ -311,6 +317,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
Expand Down
43 changes: 37 additions & 6 deletions flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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 =
Expand Down Expand Up @@ -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<>();
Expand Down Expand Up @@ -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 =
Expand Down
Loading