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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
