[Feature] Use descriptive thread names for Java and Python async executors

[wenjin272]

Search before asking

• I searched in the issues and found nothing similar.

Description

Flink Agents currently relies on the default thread names for both Java and Python asynchronous execution:

• The JDK 21 Java implementation creates its executor with Executors.newFixedThreadPool(numAsyncThreads), so worker threads have generic names such as pool-N-thread-M.
• The Python implementation creates a ThreadPoolExecutor without thread_name_prefix, so worker threads have generic names such as ThreadPoolExecutor-N_M.

These names make it difficult to identify Flink Agents async workers in TaskManager thread dumps, profiler output, and monitoring tools. A TaskManager process may contain many unrelated executor pools, and the current names do not indicate whether a thread belongs to the Flink Agents Java async executor, the Python async executor, or another component. Idle workers are especially difficult to attribute because their stacks contain only generic ThreadPoolExecutor frames.

Both executors should use descriptive, collision-resistant thread names. The naming convention should:

• clearly identify the thread as a Flink Agents async worker;
• distinguish Java and Python executor threads;
• distinguish different executor instances and workers within the same process; and
• remain useful in thread dumps and profiler output without changing execution behavior.

For example, names could follow conventions similar to:

• Java: flink-agents-java-async-<pool-id>-thread-<worker-id>
• Python: flink-agents-python-async-<pool-id>_<worker-id>

The exact convention can be decided during implementation. Java can provide a custom ThreadFactory when constructing the fixed thread pool, while Python can provide a unique thread_name_prefix to ThreadPoolExecutor.

Tests should verify that tasks submitted to each executor run on threads with the expected descriptive prefix and that names remain distinct across multiple executor instances.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[Zhuoxi2000]

I'd like to take this one. My plan is to name the async worker threads in both runtimes so they’re easier to identify in thread dumps and profiler output.

For Java, I’ll add a ThreadFactory and use names like flink-agents-async-<n>, keeping the current daemon behavior and pool sizing unchanged. For Python, I’ll set thread_name_prefix="flink-agents-async" on the ThreadPoolExecutor.

Happy to use a different naming scheme if you have a preference.