You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Savonitar (via GitHub)" <gi...@apache.org> on 2026/07/31 14:44:54 UTC
[PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Savonitar opened a new pull request, #28857:
URL: https://github.com/apache/flink/pull/28857
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## 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:
1. **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.
2. **Source coordinator worker executor** (`SourceCoordinatorContext`): `callAsync` work (e.g. periodic file-split scans) still logged without `flink-job-id`.
3. **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?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` and replace the placeholder in the "Generated-by"
line with the tool name and version. Otherwise remove the "Generated-by" line.
See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed. See
AGENTS.md for the full guidance.
-->
- [ ] Yes (please specify the tool below)
Generated-by: Claude Opus 4.8 (1M context) <no...@anthropic.com>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on PR #28857:
URL: https://github.com/apache/flink/pull/28857#issuecomment-5191405573
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719528387
##########
flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java:
##########
@@ -123,6 +124,42 @@ void testCallAsyncExceptionFailsJob() throws Exception {
"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.ofSeconds(10L),
Review Comment:
Updated to 5 minutes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on PR #28857:
URL: https://github.com/apache/flink/pull/28857#issuecomment-5193942931
thanks for review and merge!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3706097154
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -149,4 +160,25 @@ public static Map<String, String> asContextData(
context.put(JOB_ID, jobID.toHexString());
return Collections.unmodifiableMap(context);
}
+
+ /**
+ * Builds a thread-name suffix identifying the given job, e.g. {@code " (job: my-job /
+ * 0123...ef)"}. Long job names are truncated. The full hex job id matches the {@link #JOB_ID}
+ * MDC value, so thread dumps correlate with log output.
+ *
+ * @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 jobName = jobInfo.getJobName();
+ if (jobName == null || jobName.isEmpty()) {
+ return " (job: " + hexJobId + ")";
+ }
+ final String truncatedJobName =
+ jobName.length() <= MAX_JOB_NAME_IN_THREAD_NAME
+ ? jobName
+ : jobName.substring(0, MAX_JOB_NAME_IN_THREAD_NAME) + "...";
Review Comment:
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
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java:
##########
@@ -153,12 +175,35 @@ public void accept(Throwable t) {
// 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());
Review Comment:
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
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
Review Comment:
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.
##########
flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java:
##########
@@ -131,6 +133,54 @@ void testJobIdLoggedByWrappingMechanism(
assertJobIDLogged(scenario, jobID -> action.accept(jobID));
}
+ @Test
+ void testJobThreadNameSuffix() {
+ JobID jobID = new JobID();
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, "my-job")))
+ .isEqualTo(" (job: my-job / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixKeepsJobNameAtMaxLength() {
+ JobID jobID = new JobID();
+ String jobNameAtCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, jobNameAtCap)))
+ .isEqualTo(" (job: " + jobNameAtCap + " / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixTruncatesLongJobNames() {
+ JobID jobID = new JobID();
+ String jobNameOverCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME + 1);
+ String truncatedJobName = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, jobNameOverCap)))
+ .isEqualTo(" (job: " + truncatedJobName + "... / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixOmitsEmptyOrNullJobName() {
Review Comment:
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
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java:
##########
@@ -75,7 +76,7 @@ public SourceCoordinatorProvider(
@Override
public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) {
- final String coordinatorThreadName = "SourceCoordinator-" + operatorName;
+ final String coordinatorThreadName = createCoordinatorThreadName(context);
Review Comment:
This change looks like potentially not backward compatible, are there any risks with changing the thread name?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java:
##########
@@ -123,6 +124,42 @@ void testCallAsyncExceptionFailsJob() throws Exception {
"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.ofSeconds(10L),
Review Comment:
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
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java:
##########
@@ -170,7 +170,8 @@ public SourceCoordinatorContext(
new ThrowableCatchingRunnable(
this::handleUncaughtExceptionFromAsyncCall, runnable));
- this.notifier = new ExecutorNotifier(workerExecutor, errorHandlingCoordinatorExecutor);
+ // Deliberately this.workerExecutor (job-scoped), not the raw constructor parameter.
Review Comment:
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);`
##########
flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java:
##########
@@ -131,6 +133,54 @@ void testJobIdLoggedByWrappingMechanism(
assertJobIDLogged(scenario, jobID -> action.accept(jobID));
}
+ @Test
+ void testJobThreadNameSuffix() {
+ JobID jobID = new JobID();
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, "my-job")))
+ .isEqualTo(" (job: my-job / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixKeepsJobNameAtMaxLength() {
+ JobID jobID = new JobID();
+ String jobNameAtCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
Review Comment:
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.
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ void testFetcherThreadCarriesJobIdInMdcAndThreadName() throws Exception {
+ final JobID jobId = new JobID();
+ final String jobName = "my-test-job";
+
+ final FetcherThreadInfo fetcherThread =
+ captureFetcherThread(new JobInfoImpl(jobId, jobName));
+
+ assertThat(fetcherThread.mdcJobId).isEqualTo(jobId.toHexString());
+ assertThat(fetcherThread.threadName)
+ .startsWith(SplitFetcherManager.THREAD_NAME_PREFIX)
+ .endsWith(" (job: " + jobName + " / " + jobId.toHexString() + ")");
+ }
+
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ 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 testFetcherThreadNameTruncatesLongJobName() throws Exception {
+ final JobID jobId = new JobID();
+ final String longJobName = "abcdefghijklmnopqrstuvwxyz0123456789-xyz";
+ assertThat(longJobName.length()).isGreaterThan(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
+
+ final FetcherThreadInfo fetcherThread =
+ captureFetcherThread(new JobInfoImpl(jobId, longJobName));
+
+ final String expectedSuffix =
+ " (job: "
+ + longJobName.substring(0, MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME)
+ + "... / "
+ + jobId.toHexString()
+ + ")";
+ assertThat(fetcherThread.threadName).endsWith(expectedSuffix);
+ }
+
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ void testFetcherThreadNameOmitsEmptyOrNullJobName() throws Exception {
+ final JobID jobId = new JobID();
+ final String suffixWithoutJobName = " (job: " + jobId.toHexString() + ")";
+
+ final FetcherThreadInfo emptyNameThread = captureFetcherThread(new JobInfoImpl(jobId, ""));
+ assertThat(emptyNameThread.threadName).endsWith(suffixWithoutJobName);
+ assertThat(emptyNameThread.mdcJobId).isEqualTo(jobId.toHexString());
Review Comment:
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
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ void testFetcherThreadCarriesJobIdInMdcAndThreadName() throws Exception {
+ final JobID jobId = new JobID();
+ final String jobName = "my-test-job";
+
+ final FetcherThreadInfo fetcherThread =
+ captureFetcherThread(new JobInfoImpl(jobId, jobName));
+
+ assertThat(fetcherThread.mdcJobId).isEqualTo(jobId.toHexString());
+ assertThat(fetcherThread.threadName)
+ .startsWith(SplitFetcherManager.THREAD_NAME_PREFIX)
+ .endsWith(" (job: " + jobName + " / " + jobId.toHexString() + ")");
+ }
+
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ 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 testFetcherThreadNameTruncatesLongJobName() throws Exception {
Review Comment:
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
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3713261848
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
Review Comment:
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. 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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan merged PR #28857:
URL: https://github.com/apache/flink/pull/28857
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on PR #28857:
URL: https://github.com/apache/flink/pull/28857#issuecomment-5168995252
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3713261848
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
Review Comment:
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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719501124
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ void testFetcherThreadCarriesJobIdInMdcAndThreadName() throws Exception {
+ final JobID jobId = new JobID();
+ final String jobName = "my-test-job";
+
+ final FetcherThreadInfo fetcherThread =
+ captureFetcherThread(new JobInfoImpl(jobId, jobName));
+
+ assertThat(fetcherThread.mdcJobId).isEqualTo(jobId.toHexString());
+ assertThat(fetcherThread.threadName)
+ .startsWith(SplitFetcherManager.THREAD_NAME_PREFIX)
+ .endsWith(" (job: " + jobName + " / " + jobId.toHexString() + ")");
+ }
+
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ 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 testFetcherThreadNameTruncatesLongJobName() throws Exception {
Review Comment:
Agreed, it is too precautious. The truncation test is deleted from here and MdcUtilsTest owns it now alone.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719638762
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -149,4 +160,25 @@ public static Map<String, String> asContextData(
context.put(JOB_ID, jobID.toHexString());
return Collections.unmodifiableMap(context);
}
+
+ /**
+ * Builds a thread-name suffix identifying the given job, e.g. {@code " (job: my-job /
+ * 0123...ef)"}. Long job names are truncated. The full hex job id matches the {@link #JOB_ID}
+ * MDC value, so thread dumps correlate with log output.
+ *
+ * @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 jobName = jobInfo.getJobName();
+ if (jobName == null || jobName.isEmpty()) {
+ return " (job: " + hexJobId + ")";
+ }
+ final String truncatedJobName =
+ jobName.length() <= MAX_JOB_NAME_IN_THREAD_NAME
+ ? jobName
+ : jobName.substring(0, MAX_JOB_NAME_IN_THREAD_NAME) + "...";
Review Comment:
Agreed, e.g. for generated SQL job names, head-only truncation may collapse distinct jobs into identical thread names. Implemented, + added your example as the regression test: two names differing only in their -v1/-v2 tail.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on PR #28857:
URL: https://github.com/apache/flink/pull/28857#issuecomment-5190643254
Hi @Izeren . Thanks for the quick review and suggestions.
I reconsidered some tests and actually found one missed test gap and covered it. Everything (addressing your feedback and a new test) is in a new commit to simplify the review process.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719496493
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ void testFetcherThreadCarriesJobIdInMdcAndThreadName() throws Exception {
+ final JobID jobId = new JobID();
+ final String jobName = "my-test-job";
+
+ final FetcherThreadInfo fetcherThread =
+ captureFetcherThread(new JobInfoImpl(jobId, jobName));
+
+ assertThat(fetcherThread.mdcJobId).isEqualTo(jobId.toHexString());
+ assertThat(fetcherThread.threadName)
+ .startsWith(SplitFetcherManager.THREAD_NAME_PREFIX)
+ .endsWith(" (job: " + jobName + " / " + jobId.toHexString() + ")");
+ }
+
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ 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 testFetcherThreadNameTruncatesLongJobName() throws Exception {
+ final JobID jobId = new JobID();
+ final String longJobName = "abcdefghijklmnopqrstuvwxyz0123456789-xyz";
+ assertThat(longJobName.length()).isGreaterThan(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
+
+ final FetcherThreadInfo fetcherThread =
+ captureFetcherThread(new JobInfoImpl(jobId, longJobName));
+
+ final String expectedSuffix =
+ " (job: "
+ + longJobName.substring(0, MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME)
+ + "... / "
+ + jobId.toHexString()
+ + ")";
+ assertThat(fetcherThread.threadName).endsWith(expectedSuffix);
+ }
+
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
+ void testFetcherThreadNameOmitsEmptyOrNullJobName() throws Exception {
+ final JobID jobId = new JobID();
+ final String suffixWithoutJobName = " (job: " + jobId.toHexString() + ")";
+
+ final FetcherThreadInfo emptyNameThread = captureFetcherThread(new JobInfoImpl(jobId, ""));
+ assertThat(emptyNameThread.threadName).endsWith(suffixWithoutJobName);
+ assertThat(emptyNameThread.mdcJobId).isEqualTo(jobId.toHexString());
Review Comment:
Blank and null job names are now covered in the MdcUtilsTest.
I parameterised tests, but some tests left out side of parametersied case intentionally, because they test different things.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28857:
URL: https://github.com/apache/flink/pull/28857#issuecomment-5144173202
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d1afc337efe22818c9b8345eca5b6ea91204eb53",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d1afc337efe22818c9b8345eca5b6ea91204eb53",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d1afc337efe22818c9b8345eca5b6ea91204eb53 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3713281516
##########
flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java:
##########
@@ -123,6 +124,42 @@ void testCallAsyncExceptionFailsJob() throws Exception {
"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.ofSeconds(10L),
Review Comment:
https://github.com/apache/flink/pull/28857#discussion_r3713261848
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719446905
##########
flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java:
##########
@@ -131,6 +133,54 @@ void testJobIdLoggedByWrappingMechanism(
assertJobIDLogged(scenario, jobID -> action.accept(jobID));
}
+ @Test
+ void testJobThreadNameSuffix() {
+ JobID jobID = new JobID();
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, "my-job")))
+ .isEqualTo(" (job: my-job / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixKeepsJobNameAtMaxLength() {
+ JobID jobID = new JobID();
+ String jobNameAtCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
Review Comment:
The at-cap case was already covered, but I replaced tests with Parametrised test to fold similar code.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719465851
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java:
##########
@@ -170,7 +170,8 @@ public SourceCoordinatorContext(
new ThrowableCatchingRunnable(
this::handleUncaughtExceptionFromAsyncCall, runnable));
- this.notifier = new ExecutorNotifier(workerExecutor, errorHandlingCoordinatorExecutor);
+ // Deliberately this.workerExecutor (job-scoped), not the raw constructor parameter.
Review Comment:
> I assume, the reason is that wrapping from above: MdcUtils.scopeToJob(jobID, workerExecutor);
Yes, correct. Updated comment to:
```
// 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 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719626895
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java:
##########
@@ -75,7 +76,7 @@ public SourceCoordinatorProvider(
@Override
public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) {
- final String coordinatorThreadName = "SourceCoordinator-" + operatorName;
+ final String coordinatorThreadName = createCoordinatorThreadName(context);
Review Comment:
I went through every consumer of the name:
1. CoordinatorExecutorThreadFactory.isCurrentThreadCoordinatorThread() compares thread identity, not the name, so the checkState guards in SourceCoordinatorContext are unaffected.
2. getCoordinatorThreadName() is used in only two places: deriving the worker pool name, and one LOG.error message.
3. Repo-wide, "SourceCoordinator-" appears three times: once building the name in the provider, and twice in its own test, both startsWith. There is no exact-match assertion anywhere.
4. No metrics or logging configuration keys off thread names.
The SourceCoordinator-<operatorName> prefix is preserved, so prefix matching keeps working for profilers and jstack greps, only an external full-string matcher could break. Coordinator thread names already contained arbitrary operator names, including spaces for SQL operators, so spaces and parentheses are not a new class of problem for parsers.
One note: SourceCoordinatorContext derives the worker pool name as getCoordinatorThreadName() + "-worker", so worker threads now read "SourceCoordinator-<op> (job: … )-worker-thread-N" , the job suffix ends in the middle instead of at the end. Cosmetic, and nothing parses it. I have plans to improve that, but it will require more changes, bloat the PR, meanwhile it looks like there is no impact.
WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719839824
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java:
##########
@@ -153,12 +175,35 @@ public void accept(Throwable t) {
// 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());
Review Comment:
The single-argument overload picks up the enrichment for free. And two-arguments overload also isn't callable here.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719827760
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java:
##########
@@ -75,7 +76,7 @@ public SourceCoordinatorProvider(
@Override
public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) {
- final String coordinatorThreadName = "SourceCoordinator-" + operatorName;
+ final String coordinatorThreadName = createCoordinatorThreadName(context);
Review Comment:
I think it is fine as is
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719451006
##########
flink-core/src/test/java/org/apache/flink/util/MdcUtilsTest.java:
##########
@@ -131,6 +133,54 @@ void testJobIdLoggedByWrappingMechanism(
assertJobIDLogged(scenario, jobID -> action.accept(jobID));
}
+ @Test
+ void testJobThreadNameSuffix() {
+ JobID jobID = new JobID();
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, "my-job")))
+ .isEqualTo(" (job: my-job / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixKeepsJobNameAtMaxLength() {
+ JobID jobID = new JobID();
+ String jobNameAtCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, jobNameAtCap)))
+ .isEqualTo(" (job: " + jobNameAtCap + " / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixTruncatesLongJobNames() {
+ JobID jobID = new JobID();
+ String jobNameOverCap = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME + 1);
+ String truncatedJobName = "n".repeat(MdcUtils.MAX_JOB_NAME_IN_THREAD_NAME);
+ assertThat(MdcUtils.jobThreadNameSuffix(new JobInfoImpl(jobID, jobNameOverCap)))
+ .isEqualTo(" (job: " + truncatedJobName + "... / " + jobID.toHexString() + ")");
+ }
+
+ @Test
+ void testJobThreadNameSuffixOmitsEmptyOrNullJobName() {
Review Comment:
Agree, it will remove near-identical code. Done (replaced with parametrised)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719810029
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -149,4 +160,25 @@ public static Map<String, String> asContextData(
context.put(JOB_ID, jobID.toHexString());
return Collections.unmodifiableMap(context);
}
+
+ /**
+ * Builds a thread-name suffix identifying the given job, e.g. {@code " (job: my-job /
+ * 0123...ef)"}. Long job names are truncated. The full hex job id matches the {@link #JOB_ID}
+ * MDC value, so thread dumps correlate with log output.
+ *
+ * @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 jobName = jobInfo.getJobName();
+ if (jobName == null || jobName.isEmpty()) {
+ return " (job: " + hexJobId + ")";
+ }
+ final String truncatedJobName =
+ jobName.length() <= MAX_JOB_NAME_IN_THREAD_NAME
+ ? jobName
+ : jobName.substring(0, MAX_JOB_NAME_IN_THREAD_NAME) + "...";
Review Comment:
Thank you!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40270][connector-base][runtime] Make source threads job-attributable via MDC propagation and thread names [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #28857:
URL: https://github.com/apache/flink/pull/28857#discussion_r3719806241
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -75,6 +83,80 @@ void testCloseFetcherWithException() throws Exception {
.hasRootCauseMessage("Artificial exception on closing the split reader.");
}
+ @Test
+ @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
Review Comment:
Hmm, I don't particularly agree with increasing timeouts "once they flake" as with large number of tests, we are bound to introduce a ton of flakiness.
Given the history of this file, I am fine with using 60s.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org