[FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names by Savonitar · Pull Request #28857 · apache/flink

Savonitar · 2026-07-31T14:44:52Z

What is the purpose of the change

On a session / multi-tenant TaskManager, source data-plane and coordinator threads carry no job identity, so their logs and thread dumps cannot be attributed to a job (source operator names collide across jobs). Building on the JobInfo API from FLINK-39776, this PR closes three gaps:

Split fetcher threads (SplitFetcherManager): the fetcher pool did not inherit the task thread's SLF4J MDC, so every connector SplitReader logged with an empty flink-job-id, and the "Source Data Fetcher for ..." thread name carried no job identity.
Source coordinator worker executor (SourceCoordinatorContext): callAsync work (e.g. periodic file-split scans) still logged without flink-job-id.
Coordinator thread name (SourceCoordinatorProvider): SourceCoordinator-<operatorName> is identical for two jobs using the same source operator name. The thread-name suffix (truncated job name + full hex job id) deliberately matches the flink-job-id MDC value, so thread dumps correlate with log output.

Brief change log

MdcUtils: new jobThreadNameSuffix(JobInfo) helper (job name capped at 32 chars, the length of the hex JobID that follows it).
SplitFetcherManager / SingleThreadFetcherManager: new constructor overload with a trailing @Nullable JobInfo, when provided, the thread factory seeds the job id into each pool thread's MDC once and appends the job suffix to the thread name. All existing constructors delegate with null and keep their exact previous behavior.
SourceCoordinatorContext: worker executor wrapped with MdcUtils.scopeToJob.
SourceCoordinatorProvider: job suffix appended to the coordinator thread name (the derived -worker pool inherits it).

Verifying this change

This change added tests:

MdcUtilsTest: suffix format, exact truncation boundary (at cap / cap + 1), empty and null job names.
SplitFetcherManagerTest: on a real fetcher thread, the MDC carries the job id and the thread name carries the exact suffix; the no-JobInfo constructors preserve the historical thread name and empty MDC.
SourceCoordinatorContextTest: callAsync callables run with the job id in the MDC.
SourceCoordinatorProviderTest: the lazily created coordinator thread's name contains the job name and id.

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 / no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

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

Yes (please specify the tool below)

Generated-by: Claude Opus 4.8 (1M context) noreply@anthropic.com

flinkbot · 2026-07-31T14:48:44Z

CI report:

53d54c1 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

…utable via MDC propagation and thread names On a shared/multi-tenant TaskManager, source fetcher and coordinator threads carried no job identity: their logs emitted with an empty flink-job-id and their thread names could not be attributed to a job. Building on the JobInfo API from FLINK-39776, the fetcher thread factory now seeds the job id into each pool thread's MDC and appends a job suffix (truncated job name + hex job id) to fetcher and coordinator thread names, and the SourceCoordinatorContext worker executor is wrapped with MdcUtils.scopeToJob so callAsync callables carry the job id as well. All existing @PublicEvolving constructors keep their exact behavior; connectors opt in via a new trailing nullable JobInfo constructor overload on SplitFetcherManager / SingleThreadFetcherManager. Generated-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Savonitar · 2026-08-03T16:24:24Z

@flinkbot run azure

Izeren

Thank you for the change @Savonitar. I have looked through and mostly LGTM, left some suggestions around tests

Izeren · 2026-08-03T16:33:14Z

+        final String fetcherThreadName = createFetcherThreadName(taskThreadName, jobInfo);
+        if (jobInfo != null) {
+            // MDC is thread-local and not inherited, so seed the job id into each pool thread.
+            final Map<String, String> jobMdcContext = MdcUtils.asContextData(jobInfo.getJobId());


Now that this PR is merged: https://github.com/apache/flink/pull/28855/changes
What do you think of extending: MdcUtils.asContextData(jobInfo.getJobId()) here to MdcUtils.asContextData(jobId, jobInformation.getJobConfiguration()). Would it increase the coverage?

If at the time of this call, registry would already be populated, then single argument is the right choice

Izeren · 2026-08-03T16:35:41Z

    }

+    @Test
+    @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)


30s feels like a sensitive timeout for CI. They can have random VM freezes that would outlast it. Not sure if we have a common guidance on this, but I would probably put something like 5-10 min for the full test suite instead.

I know different committers prefer different options here , e.g. some start with a minimal timeout and adjust when it flakes, and that's what I did initially (also this class already uses Timeout with 30 in another test). But on the "common guidance" question: we actually have one "Avoid timeouts in JUnit tests" recommends no local timeouts at all, relying on the CI watchdog.

And this class already went through the adjustment once in FLINK-39919 it was raised from 30s to 60s after flakyness on CI. So I went with that proven pattern: dropped @Timeout from both new tests and raised the in-body wait to 60s , the same value FLINK-39919 validated for this class.

Izeren · 2026-08-03T17:00:17Z

+        final String truncatedJobName =
+                jobName.length() <= MAX_JOB_NAME_IN_THREAD_NAME
+                        ? jobName
+                        : jobName.substring(0, MAX_JOB_NAME_IN_THREAD_NAME) + "...";


Maybe would be better to to leave some amount of "last" characters. For example, if you have:
my very long job name v1
my very long job name v2

it is more helpful to see:
my ver...me v1, my ver...me v2, than generic my very long na....

Don't know which specific defaults to use, maybe ~20 from start and ~9 from end

Izeren · 2026-08-03T17:02:52Z

+    @Test
+    void testJobThreadNameSuffixKeepsJobNameAtMaxLength() {
+        JobID jobID = new JobID();
+        String jobNameAtCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);


Are we testing that at the edge, job name is preserved?

I would suggest to swap these tests for parametrised case with clear input/output arguments.

Izeren · 2026-08-03T17:05:38Z

+    }
+
+    @Test
+    void testJobThreadNameSuffixOmitsEmptyOrNullJobName() {


Could also be a part of parametrised test, actually. On second thought, we could feed in JobInfoImpl as a source as our target assert is the suffix

Izeren · 2026-08-03T17:08:27Z

                                        this::handleUncaughtExceptionFromAsyncCall, runnable));

-        this.notifier = new ExecutorNotifier(workerExecutor, errorHandlingCoordinatorExecutor);
+        // Deliberately this.workerExecutor (job-scoped), not the raw constructor parameter.


This comment explains "what". Could you please add explanation "why" is it important and what would break otherwise? I assume, the reason is that wrapping from above: MdcUtils.scopeToJob(jobID, workerExecutor);

Izeren · 2026-08-03T17:11:10Z

+
+        final FetcherThreadInfo emptyNameThread = captureFetcherThread(new JobInfoImpl(jobId, ""));
+        assertThat(emptyNameThread.threadName).endsWith(suffixWithoutJobName);
+        assertThat(emptyNameThread.mdcJobId).isEqualTo(jobId.toHexString());


Do we need to repeat this assertion for null case too? Also, maybe it would be worth to parametrise the test rather than doing the same assertions

Izeren · 2026-08-03T17:13:08Z

+
+    @Test
+    @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+    void testFetcherThreadNameTruncatesLongJobName() throws Exception {


Do we need to test the truncation logic twice in MdcUtils + here? I think we only need to test that MdcUtils have been invoked from capture method. Though it might be tricky as they are static

Izeren · 2026-08-03T17:15:06Z

    @Override
    public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) {
-        final String coordinatorThreadName = "SourceCoordinator-" + operatorName;
+        final String coordinatorThreadName = createCoordinatorThreadName(context);


This change looks like potentially not backward compatible, are there any risks with changing the thread name?

Izeren · 2026-08-03T17:16:41Z

+            coordinator.start();
+            CommonTestUtils.waitUtil(
+                    () -> findCoordinatorThread(jobInfo) != null,
+                    Duration.ofSeconds(10L),


Why specifically 10s here? Most of the test have timeouts of 30s. Same argument about risk of flakiness applies. I would suggest to use indefinite wait and test timeout instead

#28857 (comment)

Savonitar force-pushed the source-job-attribution branch from d1afc33 to 53d54c1 Compare July 31, 2026 14:54

Savonitar marked this pull request as ready for review July 31, 2026 15:32

Izeren reviewed Aug 3, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Aug 4, 2026

Uh oh!

Conversation

Savonitar 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!

Savonitar commented Aug 3, 2026

Uh oh!

Izeren 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!

Savonitar Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Savonitar commented Jul 31, 2026 •

edited

Loading

flinkbot commented Jul 31, 2026 •

edited

Loading

Savonitar Aug 4, 2026 •

edited

Loading