[FLINK-40208] Add JobMdcRegistry for config-driven MDC enrichment by Izeren · Pull Request #28855 · apache/flink

Izeren · 2026-07-31T10:32:00Z

What is the purpose of the change

Add JobMdcRegistry, a process-wide registry that maps JobID to an enriched MDC context derived from job configuration. This enables operators to surface custom job config values (e.g. org ID, environment) in Flink's JVM logs via MDC.

Brief change log

Add MdcOptions with mdc.job-configuration-to-mdc-keys (Map<String,String>) config option
Add JobMdcRegistry: static registry populated in Dispatcher and TaskExecutor
Update MdcUtils.asContextData(JobID) to consult the registry before falling back to the plain job-id singleton
Add JobIDLoggingITCase.testEnrichedMdcLogging to verify enriched keys appear in log output

Verifying this change

This change added tests and can be verified as follows:

Added JobMdcRegistryTest covering register, overwrite, unregister, and isolation between jobs
Extended MdcUtilsTest with parameterized tests for config extraction and registry-first lookup
Added JobIDLoggingITCase.testEnrichedMdcLogging as an end-to-end integration test

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes (MdcOptions is @PublicEvolving)
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? JavaDocs + config docs auto-generated from ConfigOption

Was generative AI tooling used to co-author this PR?

Yes (Claude Code)

Generated-by: Claude Code

Introduce JobMdcRegistry, a process-wide registry mapping JobID to an enriched MDC context built from job configuration. Add MdcOptions with the mdc.job-configuration-to-mdc-keys config option (@PublicEvolving). Wire in mdc enrichment on the job/task submission paths. Generated-by: Claude Code

Generated-by: Claude Code (cherry picked from commit ef6633f)

flinkbot · 2026-07-31T10:40:25Z

CI report:

4e961af Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Izeren · 2026-07-31T13:10:02Z

@rkhachatryan, thank you for the review before. This time CI is green

rkhachatryan · 2026-07-31T13:46:18Z

Merged.
Thanks for the effort!

wenshao

Reviewed — no blockers. Suggestions are inline.

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

wenshao · 2026-08-01T19:54:39Z

+title: "Logging Context (MDC)"
+weight: 7
+type: docs


[Suggestion] Hugo weight: 7 collides with sibling docs/content/docs/ops/events.md (also weight: 7), making their nav ordering non-deterministic. — Concrete cost: Hugo falls back to alphabetical tie-breaking between "Events" and "Logging Context (MDC)", so the sidebar position becomes an accident of the title rather than an editorial choice. The same collision exists in docs/content.zh/docs/ops/logging_context.md.

Suggested change

title: "Logging Context (MDC)"

weight: 7

type: docs

title: "Logging Context (MDC)"

weight: 8

type: docs

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

This looks minor, but I might include it if I do any more cleanup

wenshao · 2026-08-01T19:54:39Z

+        for (Map.Entry<String, String> entry : mdcKeyMapping.entrySet()) {
+            final String value = jobConfiguration.getString(entry.getKey(), null);
+            if (value != null && !value.isBlank()) {
+                context.put(entry.getValue(), value);


[Suggestion] The mapping target (MDC key name, entry.getValue()) is not validated for blank/empty, while the mapping source value is explicitly checked for blankness — an asymmetry that allows an empty or whitespace-only MDC key to be published. — Failure scenario: an operator configures mdc.job-configuration-to-mdc-keys: {my.key: ""} in config.yaml. The code puts an entry with an empty-string key into the context map, which is then set on every thread's MDC for that job. In Log4j2 JSON layout this produces "": "some-value" in every log record; in pattern layout, %X{} with an empty key has implementation-dependent behavior.

Suggested change

for (Map.Entry<String, String> entry : mdcKeyMapping.entrySet()) {

final String value = jobConfiguration.getString(entry.getKey(), null);

if (value != null && !value.isBlank()) {

context.put(entry.getValue(), value);

for (Map.Entry<String, String> entry : mdcKeyMapping.entrySet()) {

final String value = jobConfiguration.getString(entry.getKey(), null);

final String mdcKey = entry.getValue();

if (value != null && !value.isBlank() && mdcKey != null && !mdcKey.isBlank()) {

context.put(mdcKey, value);

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

The asymmetry here comes from the fact that key maybe empty until upstream service submitting the job is not providing the value in config. Key side acts as a "switch" for configs that we might expect. Empty value though would be a deliberate misconfiguration. So I'd not complicate the code here with extra check

wenshao · 2026-08-01T19:54:39Z

+    public static void registerOrClear(final JobID jobID, final Configuration jobConfiguration) {
+        final Map<String, String> context = MdcUtils.asContextData(jobID, jobConfiguration);
+        if (context.size() > 1) {
+            REGISTRY.put(jobID, context);
+        } else {
+            unregister(jobID);
+        }


[Suggestion] No diagnostic logging at any level when MDC enrichment is configured but resolves to zero entries, making misconfiguration invisible at runtime. — Failure scenario: an operator configures mdc.job-configuration-to-mdc-keys: pipeline.name: pipeline-name but the job is submitted without -Dpipeline.name=.... Every mapped key is silently skipped, registerOrClear calls unregister, and all log records carry only flink-job-id. At 3 AM, the oncall engineer sees no error, no warning, and no debug message explaining why the expected pipeline-name field is absent from every log record.

Consider adding a debug-level log that reports the configured mapping size and the resolved entry count, e.g.:

if (LOG.isDebugEnabled()) { LOG.debug("MDC enrichment for job {}: {} configured key(s), {} resolved entry(ies).", jobID, mdcKeyMapping.size(), context.size() - 1); }

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

Debug logs are normally not enabled in prod. Presence of the field is a signal on its own to confirm whether the feature is working. More realistic scenario is an alert setup for "missing" field and oncall engineer checking what has changed that led to field disappearing (and end to end tests, to catch such regressions before prod)

wenshao · 2026-08-01T19:54:39Z

+        JobInformation jobInformation = null;
+        try {
+            jobInformation = tdd.getJobInformation();
+        } catch (IllegalStateException ignored) {
+            // Expected when job information is offloaded to blob storage and not yet loaded.


[Suggestion] The new early-deserialization fallback for offloaded job information (the IllegalStateException catch and the jobInformation == null ternary) has no test coverage. — Failure scenario: TaskExecutorSubmissionTest.testJobMdcContextRegisteredOnSubmitTask always constructs a TDD with inline job information, so tdd.getJobInformation() never throws IllegalStateException. If a future refactor removes the catch block or the null-guarded ternary, tasks with job information offloaded to blob storage (the production path for large job graphs) would fail submission with an unhandled IllegalStateException, and no test would catch the regression.

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

Will have a look

wenshao · 2026-08-01T19:54:39Z

+        JobMdcRegistry.registerOrClear(jobId, recoveredJob.getJobConfiguration());
        try (MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobId))) {


[Suggestion] The Dispatcher recovery path (recoverJob) MDC registration and its failure-path cleanup (unregister in the catch block) have no test coverage; DispatcherTest only tests the submission path. — Failure scenario: if the registerOrClear call in recoverJob were removed during a refactor, jobs recovered after a JobManager failover would silently lose enriched MDC context — all log lines for recovered jobs would carry only flink-job-id despite the job having MDC mappings configured. No existing test would fail.

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

Will have a look

wenshao · 2026-08-01T19:54:39Z

        fileMergingManager.releaseMergingSnapshotManagerForJob(jobId);
+        JobMdcRegistry.unregister(jobId);


[Suggestion] The TaskExecutor-side cleanup (JobMdcRegistry.unregister in the job release method) has no test; TaskExecutorSubmissionTest only verifies registration, not deregistration on release. — Failure scenario: if the unregister call were removed, every completed job's MDC entry would remain in the static ConcurrentHashMap for the lifetime of the TaskManager process. On a long-running TaskManager processing hundreds of short-lived jobs, this is an unbounded memory leak in a process-wide static registry. The Dispatcher-side cleanup is tested (testJobMdcContextRegisteredOnSubmissionAndClearedOnTermination), but the TM-side equivalent is not.

— qwen3.8-max-preview via Qwen Code /review (v0.21.2)

Will have a look

Izeren · 2026-08-04T09:38:42Z

@wenshao, I have added a new PR with some of the follow ups: #28917

Izeren added 2 commits July 31, 2026 11:30

[FLINK-40208] Add Logging Context (MDC) operational docs

4e961af

Generated-by: Claude Code (cherry picked from commit ef6633f)

Izeren changed the title ~~Flink 40208/job mdc registry~~ [FLINK-40208] Add JobMdcRegistry for config-driven MDC enrichment Jul 31, 2026

rkhachatryan self-requested a review July 31, 2026 13:44

rkhachatryan approved these changes Jul 31, 2026

View reviewed changes

rkhachatryan merged commit 2779964 into apache:master Jul 31, 2026
1 check passed

wenshao reviewed Aug 1, 2026

View reviewed changes

		JobMdcRegistry.registerOrClear(jobId, recoveredJob.getJobConfiguration());
		try (MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobId))) {

		fileMergingManager.releaseMergingSnapshotManagerForJob(jobId);
		JobMdcRegistry.unregister(jobId);

Uh oh!

Conversation

Izeren commented Jul 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Jul 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Izeren commented Jul 31, 2026

Uh oh!

Uh oh!

rkhachatryan commented Jul 31, 2026

Uh oh!

wenshao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Izeren commented Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Izeren commented Jul 31, 2026 •

edited

Loading

flinkbot commented Jul 31, 2026 •

edited

Loading