This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 087caa46 [runtime][python] Use descriptive thread names for the async
executors (#1028)
087caa46 is described below
commit 087caa460d596853a8141f33bbb534186de405bf
Author: Edson <[email protected]>
AuthorDate: Fri Aug 21 04:15:58 2026 -0400
[runtime][python] Use descriptive thread names for the async executors
(#1028)
Generated-by: Claude Fable 5
---
.../flink_agents/runtime/flink_runner_context.py | 14 +++-
.../runtime/tests/test_async_thread_pool.py | 46 ++++++++++++
.../runtime/async/AsyncExecutorThreadFactory.java | 47 ++++++++++++
.../runtime/async/ContinuationActionExecutor.java | 2 +-
.../async/AsyncExecutorThreadFactoryTest.java | 83 ++++++++++++++++++++++
5 files changed, 190 insertions(+), 2 deletions(-)
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index 95539084..9f6ce590 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -15,6 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import itertools
import json
import logging
import os
@@ -881,14 +882,25 @@ def close_flink_runner_context(
ctx.close()
+_ASYNC_POOL_ID = itertools.count(1)
+"""Process-unique pool ids keeping multiple async executors distinguishable."""
+
+
def create_async_thread_pool(max_workers: int | None) -> ThreadPoolExecutor:
"""Used to create a thread pool to execute asynchronous
code block in action.
+
+ Worker threads are named
``flink-agents-python-async-<pool-id>_<worker-id>``
+ (the default ``ThreadPoolExecutor-N_M`` names make Flink Agents workers
hard
+ to attribute in TaskManager thread dumps and profiler output).
"""
logging.info(
f"Initialize fixed thread pool for async task with {max_workers}
threads"
)
- return ThreadPoolExecutor(max_workers=max_workers or os.cpu_count() * 2)
+ return ThreadPoolExecutor(
+ max_workers=max_workers or os.cpu_count() * 2,
+ thread_name_prefix=f"flink-agents-python-async-{next(_ASYNC_POOL_ID)}",
+ )
def close_async_thread_pool(executor: ThreadPoolExecutor) -> None:
diff --git a/python/flink_agents/runtime/tests/test_async_thread_pool.py
b/python/flink_agents/runtime/tests/test_async_thread_pool.py
new file mode 100644
index 00000000..0be6f0b3
--- /dev/null
+++ b/python/flink_agents/runtime/tests/test_async_thread_pool.py
@@ -0,0 +1,46 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import re
+import threading
+
+from flink_agents.runtime.flink_runner_context import (
+ close_async_thread_pool,
+ create_async_thread_pool,
+)
+
+
+def test_async_workers_carry_descriptive_names() -> None:
+ pool = create_async_thread_pool(2)
+ try:
+ name = pool.submit(lambda: threading.current_thread().name).result()
+ # ThreadPoolExecutor appends _<worker-id> to the prefix.
+ assert re.fullmatch(r"flink-agents-python-async-\d+_\d+", name), name
+ finally:
+ close_async_thread_pool(pool)
+
+
+def test_pool_prefixes_distinct_across_instances() -> None:
+ first = create_async_thread_pool(1)
+ second = create_async_thread_pool(1)
+ try:
+ first_name = first.submit(lambda:
threading.current_thread().name).result()
+ second_name = second.submit(lambda:
threading.current_thread().name).result()
+ assert first_name.rsplit("_", 1)[0] != second_name.rsplit("_", 1)[0]
+ finally:
+ close_async_thread_pool(first)
+ close_async_thread_pool(second)
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/async/AsyncExecutorThreadFactory.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/async/AsyncExecutorThreadFactory.java
new file mode 100644
index 00000000..788a9c3e
--- /dev/null
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/async/AsyncExecutorThreadFactory.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.async;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * {@link ThreadFactory} for the Flink Agents Java async executor, producing
descriptive,
+ * collision-resistant thread names of the form {@code
flink-agents-java-async-pool-N-thread-M}.
+ *
+ * <p>Default executor names such as {@code pool-N-thread-M} make Flink Agents
async workers hard to
+ * attribute in TaskManager thread dumps and profiler output, where many
unrelated pools coexist.
+ *
+ * <p>Thread creation is delegated to {@link
Executors#defaultThreadFactory()}, which normalizes
+ * daemon status and priority regardless of the calling thread (a directly
constructed {@code new
+ * Thread(...)} would inherit both from it). This factory only prepends the
{@code
+ * flink-agents-java-async-} prefix to the delegate's pool- and
worker-numbered name.
+ */
+public final class AsyncExecutorThreadFactory implements ThreadFactory {
+
+ private static final String NAME_PREFIX = "flink-agents-java-async-";
+
+ private final ThreadFactory delegate = Executors.defaultThreadFactory();
+
+ @Override
+ public Thread newThread(Runnable runnable) {
+ Thread thread = delegate.newThread(runnable);
+ thread.setName(NAME_PREFIX + thread.getName());
+ return thread;
+ }
+}
diff --git
a/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
b/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
index 4dc30d81..695d5345 100644
---
a/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
+++
b/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
@@ -43,7 +43,7 @@ public class ContinuationActionExecutor {
public ContinuationActionExecutor(int numAsyncThreads) {
LOG.info("Initialize fixed thread pool for async task with {}
threads", numAsyncThreads);
this.asyncExecutor =
- Executors.newFixedThreadPool(numAsyncThreads);
+ Executors.newFixedThreadPool(numAsyncThreads, new
AsyncExecutorThreadFactory());
}
/**
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/async/AsyncExecutorThreadFactoryTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/async/AsyncExecutorThreadFactoryTest.java
new file mode 100644
index 00000000..125d1812
--- /dev/null
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/async/AsyncExecutorThreadFactoryTest.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.async;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link AsyncExecutorThreadFactory} thread naming. */
+class AsyncExecutorThreadFactoryTest {
+
+ @Test
+ @DisplayName("Threads carry the descriptive flink-agents-java-async name")
+ void testThreadNamesCarryDescriptivePrefix() throws Exception {
+ ExecutorService executor =
+ Executors.newFixedThreadPool(2, new
AsyncExecutorThreadFactory());
+ try {
+ String name = executor.submit(() ->
Thread.currentThread().getName()).get();
+
assertThat(name).matches("flink-agents-java-async-pool-\\d+-thread-\\d+");
+ } finally {
+ executor.shutdownNow();
+ executor.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ @DisplayName("Distinct factories produce distinct pool ids, distinct
workers distinct ids")
+ void testNamesDistinctAcrossPoolsAndWorkers() {
+ AsyncExecutorThreadFactory first = new AsyncExecutorThreadFactory();
+ AsyncExecutorThreadFactory second = new AsyncExecutorThreadFactory();
+
+ String firstPoolWorker1 = first.newThread(() -> {}).getName();
+ String firstPoolWorker2 = first.newThread(() -> {}).getName();
+ String secondPoolWorker1 = second.newThread(() -> {}).getName();
+
+ assertThat(firstPoolWorker1).isNotEqualTo(firstPoolWorker2);
+ assertThat(firstPoolWorker1).isNotEqualTo(secondPoolWorker1);
+ // Pool segment differs between factories.
+ String firstPool = firstPoolWorker1.replaceAll("-thread-\\d+$", "");
+ String secondPool = secondPoolWorker1.replaceAll("-thread-\\d+$", "");
+ assertThat(firstPool).isNotEqualTo(secondPool);
+ }
+
+ @Test
+ @DisplayName(
+ "Daemon status and priority are normalized like the default
factory, not inherited")
+ void testDaemonStatusAndPriorityNormalizedNotInherited() throws Exception {
+ // Create workers from a daemon, max-priority thread: a plain new
Thread(...) would
+ // inherit both attributes, while the default-factory delegate
normalizes them.
+ AtomicReference<Thread> created = new AtomicReference<>();
+ Thread creator =
+ new Thread(() -> created.set(new
AsyncExecutorThreadFactory().newThread(() -> {})));
+ creator.setDaemon(true);
+ creator.setPriority(Thread.MAX_PRIORITY);
+ creator.start();
+ creator.join(5000);
+
+ assertThat(created.get()).isNotNull();
+ assertThat(created.get().isDaemon()).isFalse();
+
assertThat(created.get().getPriority()).isEqualTo(Thread.NORM_PRIORITY);
+ }
+}
