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