[runtime][python] Use descriptive thread names for the async executors - #1028
Conversation
github-actions
Bot
added
doc-not-needed
| this.namePrefix = "flink-agents-java-async-" + POOL_ID.incrementAndGet() + "-thread-"; | ||
| } | ||
|
|
||
| @Override |
There was a problem hiding this comment.
The PR states that only the thread name changes, but new Thread(...) inherits the daemon status and priority of the calling thread, unlike the previous default thread factory, which normalizes them to non-daemon and NORM_PRIORITY. Could we delegate thread creation to Executors.defaultThreadFactory() to preserve the existing execution semantics?
There was a problem hiding this comment.
Good catch switched to Executors.defaultThreadFactory() so daemon status and priority keep the JDK defaults, and only the name is changed. Added a regression test for that as well.
| */ | ||
| public final class AsyncExecutorThreadFactory implements ThreadFactory { | ||
|
|
||
| private static final AtomicInteger POOL_ID = new AtomicInteger(); |
There was a problem hiding this comment.
If we use Executors.defaultThreadFactory(), the generated thread name already contains the pool and worker IDs (pool-N-thread-M). Could we simply prepend flink-agents-java-async- to that name instead of maintaining separate POOL_ID and workerId counters? The resulting name would be flink-agents-java-async-pool-N-thread-M.
There was a problem hiding this comment.
Done thread names now look like flink-agents-java-async-pool-N-thread-M, and the extra counters are gone.
…lt thread factory
…names # Conflicts: # python/flink_agents/runtime/flink_runner_context.py
wenjin272
left a comment
wenjin272
left a comment
There was a problem hiding this comment.
Thanks for addressing my comments. LGTM
2 participants
Linked issue: #1013
Purpose of change
Both async executors relied on default thread names (
pool-N-thread-Mfrom the JDK 21Executors.newFixedThreadPool,ThreadPoolExecutor-N_Mfrom Python), which make Flink Agents async workers hard to attribute in TaskManager thread dumps, profiler output, and monitoring tools — especially idle workers, whose stacks contain only generic frames.Following the naming convention suggested in the issue:
AsyncExecutorThreadFactorynames workersflink-agents-java-async-<pool-id>-thread-<worker-id>, with a process-unique pool id so multiple executor instances stay distinguishable. The factory lives in the JDK-agnosticsrc/main/javasource set (the JDK 21ContinuationActionExecutorconsumes it), so it is unit-testable on every JDK. Only the name changes — daemon status and priority follow the default factory behavior.create_async_thread_poolpasses a process-uniquethread_name_prefix, yieldingflink-agents-python-async-<pool-id>_<worker-id>viaThreadPoolExecutor's standard prefix mechanism.Tests
AsyncExecutorThreadFactoryTest— tasks submitted to an executor built with the factory run on threads matchingflink-agents-java-async-\d+-thread-\d+; names are distinct across workers and across factory instances; daemon status and priority are unchanged from the default factory.test_async_thread_pool.py— submitted tasks observe names matchingflink-agents-python-async-\d+_\d+, and two pools produce distinct prefixes.API
None; thread naming only, no execution-behavior change.
Documentation
doc-neededdoc-not-neededdoc-includedWas this patch authored or co-authored using generative AI tooling?
If yes, include a
Generated-by: <tool name and version> (<model name and version>)line, for exampleGenerated-by: Claude Code 2.1.226 (Claude Opus 4.6), in the commit message so it reaches Git history. Repeat the same line here for reviewer visibility. See the ASF generative tooling guidance.Generated-by: Claude Fable 5