This is an automated email from the ASF dual-hosted git repository.
rkhachatryan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5eb36d43879 [FLINK-40270][connector-base][runtime] Make source threads
job-attributable via MDC propagation and thread names
5eb36d43879 is described below
commit 5eb36d438796ff7da2ef0a4608c216f9aad0694c
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Fri Jul 31 15:29:27 2026 +0200
[FLINK-40270][connector-base][runtime] Make source threads job-attributable
via MDC propagation and thread names
Source split-fetcher threads previously carried no job identity, so on a
shared TaskManager their logs and thread dumps could not be traced back to the
job that owned them. This adds the job id into each fetcher pool thread's MDC
and appends a truncated job-name/job-id suffix to the fetcher thread name,
making both logs and thread dumps attributable per job.
---
.../SingleThreadMultiplexSourceReaderBase.java | 21 +++-
.../reader/fetcher/SingleThreadFetcherManager.java | 22 ++++
.../source/reader/fetcher/SplitFetcherManager.java | 51 +++++++-
.../SingleThreadMultiplexSourceReaderBaseTest.java | 129 +++++++++++++++++++++
.../reader/fetcher/SplitFetcherManagerTest.java | 108 +++++++++++++++++
.../main/java/org/apache/flink/util/MdcUtils.java | 57 +++++++++
.../java/org/apache/flink/util/MdcUtilsTest.java | 69 ++++++++++-
.../coordinator/SourceCoordinatorContext.java | 7 +-
.../coordinator/SourceCoordinatorProvider.java | 18 ++-
.../coordinator/SourceCoordinatorContextTest.java | 42 +++++++
.../coordinator/SourceCoordinatorProviderTest.java | 37 ++++++
11 files changed, 552 insertions(+), 9 deletions(-)
diff --git
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java
index 023a7d0c50d..36f2762cc4a 100644
---
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java
+++
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.base.source.reader;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.SourceSplit;
@@ -75,7 +76,8 @@ public abstract class SingleThreadMultiplexSourceReaderBase<
Configuration config,
SourceReaderContext context) {
super(
- new SingleThreadFetcherManager<>(splitReaderSupplier, config),
+ new SingleThreadFetcherManager<>(
+ splitReaderSupplier, config, (ignore) -> {},
getJobInfoOrNull(context)),
recordEmitter,
config,
context);
@@ -93,7 +95,8 @@ public abstract class SingleThreadMultiplexSourceReaderBase<
SourceReaderContext context,
@Nullable RateLimiterStrategy<SplitT> rateLimiterStrategy) {
super(
- new SingleThreadFetcherManager<>(splitReaderSupplier, config),
+ new SingleThreadFetcherManager<>(
+ splitReaderSupplier, config, (ignore) -> {},
getJobInfoOrNull(context)),
recordEmitter,
null,
config,
@@ -149,4 +152,18 @@ public abstract class
SingleThreadMultiplexSourceReaderBase<
context,
rateLimiterStrategy);
}
+
+ /**
+ * Returns the {@link JobInfo} of the given context, or {@code null} if
the context (e.g. an
+ * older runtime or a test double) does not implement {@link
SourceReaderContext#getJobInfo()},
+ * which throws {@link UnsupportedOperationException} by default.
+ */
+ @Nullable
+ private static JobInfo getJobInfoOrNull(SourceReaderContext context) {
+ try {
+ return context.getJobInfo();
+ } catch (UnsupportedOperationException e) {
+ return null;
+ }
+ }
}
diff --git
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
index 642d4ae495d..adb903cbcd6 100644
---
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
+++
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
@@ -19,10 +19,13 @@
package org.apache.flink.connector.base.source.reader.fetcher;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import javax.annotation.Nullable;
+
import java.util.Collection;
import java.util.List;
import java.util.function.Consumer;
@@ -78,6 +81,25 @@ public class SingleThreadFetcherManager<E, SplitT extends
SourceSplit>
super(splitReaderSupplier, configuration, splitFinishedHook);
}
+ /**
+ * Creates a new SplitFetcherManager with a single I/O thread.
+ *
+ * @param splitReaderSupplier The factory for the split reader that
connects to the source
+ * system.
+ * @param configuration The configuration to create the fetcher manager.
+ * @param splitFinishedHook Hook for handling finished splits in split
fetchers
+ * @param jobInfo The job this fetcher manager belongs to, or {@code null}
if unknown. See
+ * {@link SplitFetcherManager#SplitFetcherManager(Supplier,
Configuration, Consumer,
+ * JobInfo)}.
+ */
+ public SingleThreadFetcherManager(
+ Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
+ Configuration configuration,
+ Consumer<Collection<String>> splitFinishedHook,
+ @Nullable JobInfo jobInfo) {
+ super(splitReaderSupplier, configuration, splitFinishedHook, jobInfo);
+ }
+
@Override
public void addSplits(List<SplitT> splitsToAdd) {
SplitFetcher<E, SplitT> fetcher = getRunningFetcher();
diff --git
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
index e101bc4861c..6ea31d2d53b 100644
---
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
+++
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.base.source.reader.fetcher;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -28,10 +29,13 @@ import
org.apache.flink.connector.base.source.reader.SourceReaderBase;
import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.util.MdcUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
@@ -126,6 +130,24 @@ public abstract class SplitFetcherManager<E, SplitT
extends SourceSplit> {
Supplier<SplitReader<E, SplitT>> splitReaderFactory,
Configuration configuration,
Consumer<Collection<String>> splitFinishedHook) {
+ this(splitReaderFactory, configuration, splitFinishedHook, null);
+ }
+
+ /**
+ * Create a split fetcher manager.
+ *
+ * @param splitReaderFactory a supplier that could be used to create split
readers.
+ * @param configuration the configuration of this fetcher manager.
+ * @param splitFinishedHook Hook for handling finished splits in split
fetchers.
+ * @param jobInfo the job this fetcher manager belongs to, or {@code null}
if unknown. When
+ * provided, fetcher threads carry the job id in their MDC ({@value
MdcUtils#JOB_ID}) and
+ * thread names, making their logs and thread dumps attributable to
the job.
+ */
+ public SplitFetcherManager(
+ Supplier<SplitReader<E, SplitT>> splitReaderFactory,
+ Configuration configuration,
+ Consumer<Collection<String>> splitFinishedHook,
+ @Nullable JobInfo jobInfo) {
this.elementsQueue =
new FutureCompletingBlockingQueue<>(
configuration.get(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY));
@@ -153,12 +175,35 @@ public abstract class SplitFetcherManager<E, SplitT
extends SourceSplit> {
// Create the executor with a thread factory that fails the source
reader if one of
// the fetcher thread exits abnormally.
final String taskThreadName = Thread.currentThread().getName();
- this.executors =
- Executors.newCachedThreadPool(
- r -> new Thread(r, THREAD_NAME_PREFIX +
taskThreadName));
+ 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());
+ this.executors =
+ Executors.newCachedThreadPool(
+ r ->
+ new Thread(
+
MdcUtils.wrapRunnable(jobMdcContext, r),
+ fetcherThreadName));
+ } else {
+ this.executors = Executors.newCachedThreadPool(r -> new Thread(r,
fetcherThreadName));
+ }
this.closed = false;
}
+ /**
+ * Builds the name shared by all fetcher threads of this manager. When the
job is known, a
+ * {@link MdcUtils#jobThreadNameSuffix(JobInfo) job suffix} is appended so
fetcher threads of
+ * different jobs are distinguishable on a shared TaskManager.
+ */
+ private static String createFetcherThreadName(
+ String taskThreadName, @Nullable JobInfo jobInfo) {
+ if (jobInfo == null) {
+ return THREAD_NAME_PREFIX + taskThreadName;
+ }
+ return THREAD_NAME_PREFIX + taskThreadName +
MdcUtils.jobThreadNameSuffix(jobInfo);
+ }
+
public abstract void addSplits(List<SplitT> splitsToAdd);
public abstract void removeSplits(List<SplitT> splitsToRemove);
diff --git
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBaseTest.java
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBaseTest.java
new file mode 100644
index 00000000000..d59433684e8
--- /dev/null
+++
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBaseTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.connector.base.source.reader;
+
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.mocks.MockSourceReader;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.util.MdcUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests the job attribution that {@link
SingleThreadMultiplexSourceReaderBase} gives the fetcher
+ * threads it creates.
+ */
+class SingleThreadMultiplexSourceReaderBaseTest {
+
+ @Test
+ void testFetcherThreadNameIdentifiesJob() throws Exception {
+ final TestingReaderContext context = new TestingReaderContext();
+
+ assertThat(fetcherThreadNameOf(context))
+ .endsWith(MdcUtils.jobThreadNameSuffix(context.getJobInfo()));
+ }
+
+ /**
+ * A context that leaves {@link SourceReaderContext#getJobInfo()} at its
throwing default must
+ * still yield a working reader, only one without a job suffix:
attribution is diagnostic only.
+ */
+ @Test
+ void testContextWithoutJobInfoYieldsUnattributedFetcherThread() throws
Exception {
+ final SourceReaderContext context =
+ new TestingReaderContext() {
+ @Override
+ public JobInfo getJobInfo() {
+ throw new UnsupportedOperationException();
+ }
+ };
+
+ assertThat(fetcherThreadNameOf(context))
+ .as("an unattributable reader must not gain a job suffix")
+ .doesNotContain(" (job: ");
+ }
+
+ /**
+ * Builds a reader over the given context and assigns it a split, so that
the fetcher thread
+ * starts and can report its own name.
+ */
+ private static String fetcherThreadNameOf(SourceReaderContext context)
throws Exception {
+ final CompletableFuture<String> fetcherThreadName = new
CompletableFuture<>();
+ try (MockSourceReader reader =
+ new MockSourceReader(
+ () -> new
ThreadNameReportingSplitReader(fetcherThreadName),
+ new Configuration(),
+ context)) {
+ reader.start();
+ reader.addSplits(Collections.singletonList(new MockSourceSplit(0,
0, 1)));
+ assertThat(fetcherThreadName)
+ .as("The fetcher thread should have started fetching.")
+ .succeedsWithin(Duration.ofSeconds(60));
+ return fetcherThreadName.get();
+ }
+ }
+
+ /** Reports the thread it is driven on, which is the fetcher thread under
test. */
+ private static final class ThreadNameReportingSplitReader
+ implements SplitReader<int[], MockSourceSplit> {
+
+ private final CompletableFuture<String> threadName;
+ private final OneShotLatch fetchBlocker = new OneShotLatch();
+
+ private ThreadNameReportingSplitReader(CompletableFuture<String>
threadName) {
+ this.threadName = threadName;
+ }
+
+ @Override
+ public RecordsWithSplitIds<int[]> fetch() {
+ threadName.complete(Thread.currentThread().getName());
+ // Stay inside fetch() until woken up, so the fetcher does not
spin on empty fetches.
+ try {
+ fetchBlocker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return new RecordsBySplits<>(Collections.emptyMap(),
Collections.emptySet());
+ }
+
+ @Override
+ public void handleSplitsChanges(SplitsChange<MockSourceSplit>
splitsChanges) {}
+
+ @Override
+ public void wakeUp() {
+ fetchBlocker.trigger();
+ }
+
+ @Override
+ public void close() {
+ fetchBlocker.trigger();
+ }
+ }
+}
diff --git
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
index 742571e37dd..cf0fce5d42c 100644
---
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
+++
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
@@ -18,6 +18,9 @@
package org.apache.flink.connector.base.source.reader.fetcher;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordsBySplits;
@@ -30,9 +33,13 @@ import
org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.util.MdcUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.slf4j.MDC;
+
+import javax.annotation.Nullable;
import java.io.IOException;
import java.time.Duration;
@@ -42,6 +49,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.apache.flink.test.util.TestUtils.waitUntil;
@@ -75,6 +83,36 @@ class SplitFetcherManagerTest {
.hasRootCauseMessage("Artificial exception on closing the
split reader.");
}
+ /**
+ * The exact suffix format is covered by {@code
MdcUtilsTest#testJobThreadNameSuffix}; this only
+ * has to show that a fetcher thread is seeded with the job MDC and named
after the job at all.
+ */
+ @Test
+ void testFetcherThreadCarriesJobIdInMdcAndThreadName() throws Exception {
+ final JobID jobId = new JobID();
+ final JobInfo jobInfo = new JobInfoImpl(jobId, "my-test-job");
+ final String taskThreadName = Thread.currentThread().getName();
+
+ final FetcherThreadInfo fetcherThread = captureFetcherThread(jobInfo);
+
+ assertThat(fetcherThread.mdcJobId).isEqualTo(jobId.toHexString());
+ assertThat(fetcherThread.threadName)
+ .startsWith(SplitFetcherManager.THREAD_NAME_PREFIX)
+ .contains(taskThreadName)
+ .endsWith(MdcUtils.jobThreadNameSuffix(jobInfo));
+ }
+
+ @Test
+ void testFetcherThreadWithoutJobInfoKeepsHistoricalNameAndNoJobIdInMdc()
throws Exception {
+ final FetcherThreadInfo fetcherThread = captureFetcherThread(null);
+
+ // Fetcher threads are named after the thread creating the manager,
i.e. this test thread.
+ assertThat(fetcherThread.threadName)
+ .isEqualTo(
+ SplitFetcherManager.THREAD_NAME_PREFIX +
Thread.currentThread().getName());
+ assertThat(fetcherThread.mdcJobId).isNull();
+ }
+
@Test
@Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
void testCloseCleansUpPreviouslyClosedFetcher() throws Exception {
@@ -243,6 +281,31 @@ class SplitFetcherManagerTest {
return fetcher;
}
+ /**
+ * Runs a fetcher for the given job identity ({@code null} exercising the
constructors without
+ * {@link JobInfo}) and returns what its fetcher thread saw. The fetcher
manager is closed again
+ * before returning.
+ */
+ private static FetcherThreadInfo captureFetcherThread(@Nullable JobInfo
jobInfo)
+ throws Exception {
+ final ThreadInfoCapturingSplitReader<Integer> reader =
+ new ThreadInfoCapturingSplitReader<>();
+ final SingleThreadFetcherManager<Integer, TestingSourceSplit>
fetcherManager =
+ jobInfo == null
+ ? new SingleThreadFetcherManager<>(() -> reader, new
Configuration())
+ : new SingleThreadFetcherManager<>(
+ () -> reader, new Configuration(), (ignore) ->
{}, jobInfo);
+ try {
+ fetcherManager.addSplits(Collections.singletonList(new
TestingSourceSplit("split-0")));
+ assertThat(reader.threadInfo)
+ .as("The fetcher thread should have started fetching.")
+ .succeedsWithin(Duration.ofSeconds(60));
+ return reader.threadInfo.get();
+ } finally {
+ fetcherManager.close(30_000L);
+ }
+ }
+
private static void drainQueue(FutureCompletingBlockingQueue<?> queue) {
//noinspection StatementWithEmptyBody
while (queue.poll() != null) {}
@@ -262,6 +325,51 @@ class SplitFetcherManagerTest {
// test mocks
// ------------------------------------------------------------------------
+ /** Thread name and {@value MdcUtils#JOB_ID} MDC value observed on a
fetcher thread. */
+ private static final class FetcherThreadInfo {
+
+ private final String threadName;
+ @Nullable private final String mdcJobId;
+
+ private FetcherThreadInfo(String threadName, @Nullable String
mdcJobId) {
+ this.threadName = threadName;
+ this.mdcJobId = mdcJobId;
+ }
+ }
+
+ /** A {@link SplitReader} that reports the fetcher thread it is executed
on. */
+ private static final class ThreadInfoCapturingSplitReader<E>
+ implements SplitReader<E, TestingSourceSplit> {
+
+ private final CompletableFuture<FetcherThreadInfo> threadInfo = new
CompletableFuture<>();
+ private final OneShotLatch fetchBlocker = new OneShotLatch();
+
+ @Override
+ public RecordsWithSplitIds<E> fetch() {
+ threadInfo.complete(
+ new FetcherThreadInfo(
+ Thread.currentThread().getName(),
MDC.get(MdcUtils.JOB_ID)));
+ // Stay inside fetch() until woken up, so the fetcher does not
spin on empty fetches.
+ try {
+ fetchBlocker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return new RecordsBySplits<>(Collections.emptyMap(),
Collections.emptySet());
+ }
+
+ @Override
+ public void handleSplitsChanges(SplitsChange<TestingSourceSplit>
splitsChanges) {}
+
+ @Override
+ public void wakeUp() {
+ fetchBlocker.trigger();
+ }
+
+ @Override
+ public void close() {}
+ }
+
private static final class AwaitingReader<E, SplitT extends SourceSplit>
implements SplitReader<E, SplitT> {
diff --git a/flink-core/src/main/java/org/apache/flink/util/MdcUtils.java
b/flink-core/src/main/java/org/apache/flink/util/MdcUtils.java
index 935dfa79505..979b8a2c943 100644
--- a/flink-core/src/main/java/org/apache/flink/util/MdcUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/MdcUtils.java
@@ -19,11 +19,14 @@
package org.apache.flink.util;
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.MDC;
+import javax.annotation.Nonnull;
+
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -39,6 +42,27 @@ public class MdcUtils {
public static final String JOB_ID = "flink-job-id";
+ /**
+ * 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.
+ */
+ private static final int MAX_JOB_NAME_IN_THREAD_NAME = 32;
+
+ /**
+ * Number of trailing job name characters kept when a job name is
truncated. Generated job names
+ * often share a long prefix and differ only near the end (e.g. {@code
...-v1} / {@code
+ * ...-v2}), so dropping the tail would make distinct jobs
indistinguishable in a thread dump.
+ */
+ private static final int TRUNCATED_JOB_NAME_TAIL_LENGTH = 9;
+
+ private static final String TRUNCATION_MARKER = "...";
+
+ /** Chosen so that a truncated name is exactly {@link
#MAX_JOB_NAME_IN_THREAD_NAME} long. */
+ private static final int TRUNCATED_JOB_NAME_HEAD_LENGTH =
+ MAX_JOB_NAME_IN_THREAD_NAME
+ - TRUNCATION_MARKER.length()
+ - TRUNCATED_JOB_NAME_TAIL_LENGTH;
+
/**
* Replace MDC contents with the provided one and return a closeable
object that can be used to
* restore the original MDC.
@@ -149,4 +173,37 @@ public class MdcUtils {
context.put(JOB_ID, jobID.toHexString());
return Collections.unmodifiableMap(context);
}
+
+ /**
+ * Build a thread-name suffix identifying the job, e.g. {@code " (job:
my-job /
+ * 2f4b0e4a9cbb223e924f1e5d9e6a7c11)"}. The job name may be truncated; the
hex job id never is,
+ * so it always matches the {@link #JOB_ID} MDC value and a thread dump
can be lined up with the
+ * logs.
+ *
+ * @param jobInfo the job meta information
+ * @return a suffix to append to a thread name
+ */
+ public static String jobThreadNameSuffix(@Nonnull JobInfo jobInfo) {
+ final String hexJobId = jobInfo.getJobId().toHexString();
+ final String rawJobName = jobInfo.getJobName();
+ final String jobName = rawJobName == null ? "" : rawJobName.strip();
+ if (jobName.isEmpty()) {
+ return " (job: " + hexJobId + ")";
+ }
+ return " (job: " + truncateJobName(jobName) + " / " + hexJobId + ")";
+ }
+
+ /**
+ * Shorten a job name to {@link #MAX_JOB_NAME_IN_THREAD_NAME} characters
by eliding its middle
+ * and keeping the last {@link #TRUNCATED_JOB_NAME_TAIL_LENGTH}, for the
reason given on that
+ * constant.
+ */
+ private static String truncateJobName(String jobName) {
+ if (jobName.length() <= MAX_JOB_NAME_IN_THREAD_NAME) {
+ return jobName;
+ }
+ return jobName.substring(0, TRUNCATED_JOB_NAME_HEAD_LENGTH)
+ + TRUNCATION_MARKER
+ + jobName.substring(jobName.length() -
TRUNCATED_JOB_NAME_TAIL_LENGTH);
+ }
}
diff --git a/flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java
b/flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java
index 117a51b74bc..8075bffd9a1 100644
--- a/flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java
+++ b/flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java
@@ -19,6 +19,8 @@
package org.apache.flink.util;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MdcOptions;
import
org.apache.flink.core.testutils.ManuallyTriggeredScheduledExecutorService;
@@ -128,7 +130,72 @@ class MdcUtilsTest {
void testJobIdLoggedByWrappingMechanism(
final String scenario, final ThrowingConsumer<JobID, Exception>
action)
throws Exception {
- assertJobIDLogged(scenario, jobID -> action.accept(jobID));
+ assertJobIDLogged(scenario, action);
+ }
+
+ /**
+ * Expected strings are literals rather than values computed from the
truncation constants:
+ * computing them would make the test agree with the production formula
even when that formula
+ * is wrong. Job names use a distinct character per position so that the
head and tail
+ * boundaries in each literal can be checked by eye against the input.
+ */
+ private static Stream<Arguments> jobThreadNameSuffixCases() {
+ final JobID jobID = new JobID();
+ final String hexJobId = jobID.toHexString();
+ final String nameAtCap = "0123456789abcdefghijABCDEFGHIJxy";
+ return Stream.of(
+ Arguments.of(
+ "short name kept verbatim",
+ new JobInfoImpl(jobID, "my-job"),
+ " (job: my-job / " + hexJobId + ")"),
+ Arguments.of(
+ "padded name stripped",
+ new JobInfoImpl(jobID, " my-job "),
+ " (job: my-job / " + hexJobId + ")"),
+ Arguments.of(
+ "name at the cap kept verbatim",
+ new JobInfoImpl(jobID, nameAtCap),
+ " (job: " + nameAtCap + " / " + hexJobId + ")"),
+ Arguments.of(
+ "one character over the cap is elided in the middle",
+ new JobInfoImpl(jobID, nameAtCap + "z"),
+ " (job: 0123456789abcdefghij...EFGHIJxyz / " +
hexJobId + ")"),
+ Arguments.of(
+ "long name keeps head and tail (v1)",
+ new JobInfoImpl(jobID,
"0123456789abcdefghijABCDEFGHIJ-job-v1"),
+ " (job: 0123456789abcdefghij...IJ-job-v1 / " +
hexJobId + ")"),
+ Arguments.of(
+ "empty name omitted",
+ new JobInfoImpl(jobID, ""),
+ " (job: " + hexJobId + ")"),
+ Arguments.of(
+ "blank name omitted",
+ new JobInfoImpl(jobID, " "),
+ " (job: " + hexJobId + ")"),
+ Arguments.of(
+ "null name omitted", nullNameJobInfo(jobID), " (job: "
+ hexJobId + ")"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("jobThreadNameSuffixCases")
+ void testJobThreadNameSuffix(
+ final String scenario, final JobInfo jobInfo, final String
expectedSuffix) {
+
assertThat(MdcUtils.jobThreadNameSuffix(jobInfo)).isEqualTo(expectedSuffix);
+ }
+
+ /** {@link JobInfoImpl} rejects a null name, so null handling needs a
hand-written one. */
+ private static JobInfo nullNameJobInfo(JobID jobID) {
+ return new JobInfo() {
+ @Override
+ public JobID getJobId() {
+ return jobID;
+ }
+
+ @Override
+ public String getJobName() {
+ return null;
+ }
+ };
}
@Test
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
index bb2d6a30050..67e20fac4b9 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
@@ -154,7 +154,7 @@ public class SourceCoordinatorContext<SplitT extends
SourceSplit>
SimpleVersionedSerializer<SplitT> splitSerializer,
SplitAssignmentTracker<SplitT> splitAssignmentTracker,
boolean supportsConcurrentExecutionAttempts) {
- this.workerExecutor = workerExecutor;
+ this.workerExecutor = MdcUtils.scopeToJob(jobID, workerExecutor);
this.coordinatorExecutor = MdcUtils.scopeToJob(jobID,
coordinatorExecutor);
this.coordinatorThreadFactory = coordinatorThreadFactory;
this.operatorCoordinatorContext = operatorCoordinatorContext;
@@ -170,7 +170,10 @@ public class SourceCoordinatorContext<SplitT extends
SourceSplit>
new ThrowableCatchingRunnable(
this::handleUncaughtExceptionFromAsyncCall, runnable));
- this.notifier = new ExecutorNotifier(workerExecutor,
errorHandlingCoordinatorExecutor);
+ // Must be the field, not the constructor parameter: the field is the
scopeToJob-wrapped
+ // executor, so the callables ExecutorNotifier schedules on it,
one-shot and periodic alike,
+ // log with the job id rather than an empty MDC.
+ this.notifier = new ExecutorNotifier(this.workerExecutor,
errorHandlingCoordinatorExecutor);
}
boolean isConcurrentExecutionAttemptsSupported() {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
index 742a0bcb563..6c4b6722663 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
@@ -25,6 +25,7 @@ import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import
org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator;
+import org.apache.flink.util.MdcUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -75,7 +76,7 @@ public class SourceCoordinatorProvider<SplitT extends
SourceSplit>
@Override
public OperatorCoordinator getCoordinator(OperatorCoordinator.Context
context) {
- final String coordinatorThreadName = "SourceCoordinator-" +
operatorName;
+ final String coordinatorThreadName =
createCoordinatorThreadName(context);
CoordinatorExecutorThreadFactory coordinatorThreadFactory =
new CoordinatorExecutorThreadFactory(coordinatorThreadName,
context);
@@ -98,6 +99,21 @@ public class SourceCoordinatorProvider<SplitT extends
SourceSplit>
coordinatorListeningID);
}
+ /**
+ * Builds the coordinator thread name. The operator name alone is
ambiguous on a shared cluster
+ * because two jobs may use identically named sources, so the job identity
is appended to make
+ * the coordinator thread (and its derived {@code -worker} pool)
attributable to a job.
+ */
+ private String createCoordinatorThreadName(OperatorCoordinator.Context
context) {
+ final String base = "SourceCoordinator-" + operatorName;
+ try {
+ return base + MdcUtils.jobThreadNameSuffix(context.getJobInfo());
+ } catch (UnsupportedOperationException e) {
+ // A custom Context may not expose job identity - fall back to the
operator-only name.
+ return base;
+ }
+ }
+
/**
* A thread factory class that provides some helper methods. Because it is
used to check the
* current thread, it is a one-off, do not use this ThreadFactory to
create multiple threads.
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
index f6db6e5bb8e..aa94c861f32 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
@@ -29,8 +29,10 @@ import
org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.runtime.source.event.AddSplitEvent;
import org.apache.flink.runtime.source.event.IsProcessingBacklogEvent;
import org.apache.flink.runtime.source.event.ReaderRegistrationEvent;
+import org.apache.flink.util.MdcUtils;
import org.junit.jupiter.api.Test;
+import org.slf4j.MDC;
import java.util.Arrays;
import java.util.Collections;
@@ -252,6 +254,46 @@ class SourceCoordinatorContextTest extends
SourceCoordinatorTestBase {
assertThat(operatorCoordinatorContext.isJobFailed()).isFalse();
}
+ @Test
+ void testCallAsyncCallableRunsWithJobIdInMdc() throws Exception {
+ final JobID jobId = new JobID();
+ final AtomicReference<String> mdcJobIdInCallable = new
AtomicReference<>();
+
+ ManuallyTriggeredScheduledExecutorService manualWorkerExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ ManuallyTriggeredScheduledExecutorService manualCoordinatorExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+
+ SourceCoordinatorContext<MockSourceSplit> testingContext =
+ new SourceCoordinatorContext<>(
+ jobId,
+ manualCoordinatorExecutor,
+ manualWorkerExecutor,
+ new
SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
+ TEST_OPERATOR_ID.toHexString(),
operatorCoordinatorContext),
+ operatorCoordinatorContext,
+ new MockSourceSplitSerializer(),
+ splitSplitAssignmentTracker,
+ false);
+
+ try {
+ // The callable runs on the worker executor, which must be
job-scoped.
+ testingContext.callAsync(
+ () -> {
+ mdcJobIdInCallable.set(MDC.get(MdcUtils.JOB_ID));
+ return null;
+ },
+ (ignored, e) -> {});
+
+ // triggerAll() runs the queued callable synchronously on this
thread.
+ manualWorkerExecutor.triggerAll();
+
+
assertThat(mdcJobIdInCallable.get()).isEqualTo(jobId.toHexString());
+ } finally {
+ testingContext.close();
+ }
+ }
+
@Test
void testSupportsIntermediateNoMoreSplits() throws Exception {
sourceReady();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
index 11fefa5b981..8be7d66559b 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
@@ -18,6 +18,7 @@ limitations under the License.
package org.apache.flink.runtime.source.coordinator;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.eventtime.WatermarkAlignmentParams;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.mocks.MockSource;
@@ -123,6 +124,42 @@ class SourceCoordinatorProviderTest {
"The job did not fail before timeout.");
}
+ @Test
+ void testCoordinatorThreadNameContainsJobIdentity() throws Exception {
+ final MockOperatorCoordinatorContext context =
+ new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SPLITS);
+ final RecreateOnResetOperatorCoordinator coordinator =
+ (RecreateOnResetOperatorCoordinator) provider.create(context);
+ final JobInfo jobInfo = context.getJobInfo();
+ try {
+ // Starting the coordinator creates the (lazily initialized)
coordinator thread.
+ coordinator.start();
+ CommonTestUtils.waitUtil(
+ () -> findCoordinatorThread(jobInfo) != null,
+ Duration.ofMinutes(5L),
+ "The coordinator thread carrying the job identity was not
found.");
+
+ final Thread coordinatorThread = findCoordinatorThread(jobInfo);
+ assertThat(coordinatorThread).isNotNull();
+ assertThat(coordinatorThread.getName())
+
.startsWith("SourceCoordinator-SourceCoordinatorProviderTest")
+ .contains(jobInfo.getJobName())
+ .contains(jobInfo.getJobId().toHexString());
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ private static Thread findCoordinatorThread(JobInfo jobInfo) {
+ for (Thread t : Thread.getAllStackTraces().keySet()) {
+ if (t.getName().startsWith("SourceCoordinator-")
+ && t.getName().contains(jobInfo.getJobId().toHexString()))
{
+ return t;
+ }
+ }
+ return null;
+ }
+
@Test
void testCoordinatorExecutorThreadFactoryNewMultipleThread() {
SourceCoordinatorProvider.CoordinatorExecutorThreadFactory
