wenshao commented on code in PR #28855:
URL: https://github.com/apache/flink/pull/28855#discussion_r3696500725
##########
docs/content/docs/ops/logging_context.md:
##########
@@ -0,0 +1,86 @@
+---
+title: "Logging Context (MDC)"
+weight: 7
+type: docs
Review Comment:
**[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`.
```suggestion
title: "Logging Context (MDC)"
weight: 8
type: docs
```
_— qwen3.8-max-preview via Qwen Code /review (v0.21.2)_
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -112,7 +115,38 @@ public static ScheduledExecutorService scopeToJob(JobID
jobID, ScheduledExecutor
return new MdcAwareScheduledExecutorService(ses, asContextData(jobID));
}
+ /**
+ * Build MDC context for a job. Consults the {@link JobMdcRegistry} for
enriched context
+ * registered where the job {@link Configuration} is available; falls back
to the plain job ID
+ * entry.
+ */
public static Map<String, String> asContextData(JobID jobID) {
+ final Map<String, String> registered = JobMdcRegistry.lookup(jobID);
+ if (registered != null) {
+ return registered;
+ }
return Collections.singletonMap(JOB_ID, jobID.toHexString());
}
+
+ /**
+ * Build MDC context from a job ID and job configuration, enriching with
context entries
+ * configured via {@link MdcOptions#JOB_CONFIGURATION_TO_MDC_KEYS}.
+ */
+ public static Map<String, String> asContextData(
+ final JobID jobID, final Configuration jobConfiguration) {
+ final Map<String, String> mdcKeyMapping =
+ 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 value = jobConfiguration.getString(entry.getKey(),
null);
+ if (value != null && !value.isBlank()) {
+ context.put(entry.getValue(), value);
Review Comment:
**[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.
```suggestion
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)_
##########
flink-core/src/main/java/org/apache/flink/util/JobMdcRegistry.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Process-wide registry mapping {@link JobID} to enriched MDC context,
populated where the job
+ * {@link Configuration} is available and consulted by {@link
MdcUtils#asContextData(JobID)}.
+ */
+@Internal
+@ThreadSafe
+public final class JobMdcRegistry {
+
+ private static final Map<JobID, Map<String, String>> REGISTRY = new
ConcurrentHashMap<>();
+
+ private JobMdcRegistry() {}
+
+ /**
+ * Registers enriched MDC context if the configuration carries any MDC key
mappings; clears any
+ * stale entry otherwise. Equivalent to {@link #unregister} when the
config is unenriched.
+ */
+ 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);
+ }
Review Comment:
**[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.:
```java
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)_
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -2078,6 +2084,7 @@ private void releaseJobResources(JobID jobId, Exception
cause) {
taskInformationCache.clearCacheForGroup(jobId);
shuffleDescriptorsCache.clearCacheForGroup(jobId);
fileMergingManager.releaseMergingSnapshotManagerForJob(jobId);
+ JobMdcRegistry.unregister(jobId);
Review Comment:
**[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)_
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -661,9 +662,20 @@ public CompletableFuture<Acknowledge> submitTask(
TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, Duration
timeout) {
final JobID jobId = tdd.getJobId();
- // todo: consider adding task info
- try (MdcCloseable ignored =
MdcUtils.withContext(MdcUtils.asContextData(jobId))) {
-
+ JobInformation jobInformation = null;
+ try {
+ jobInformation = tdd.getJobInformation();
+ } catch (IllegalStateException ignored) {
+ // Expected when job information is offloaded to blob storage and
not yet loaded.
Review Comment:
**[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)_
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -572,6 +573,7 @@ private void runRecoveredJob(
initJobClientExpiredTime(recoveredJob);
final JobID jobId = recoveredJob.getJobID();
+ JobMdcRegistry.registerOrClear(jobId,
recoveredJob.getJobConfiguration());
try (MdcCloseable ignored =
MdcUtils.withContext(MdcUtils.asContextData(jobId))) {
Review Comment:
**[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)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
