This is an automated email from the ASF dual-hosted git repository.
jtuglu1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new dbe1d23ee03 fix: ensure supervisor worker threadpool is reapable
(#19738)
dbe1d23ee03 is described below
commit dbe1d23ee03a8ee96cbe2b2b100912a680445b66
Author: jtuglu1 <[email protected]>
AuthorDate: Thu Jul 23 15:09:35 2026 -0700
fix: ensure supervisor worker threadpool is reapable (#19738)
Currently, we permit unbounded native thread allocation based on user input
(taskCountMax):
https://github.com/apache/druid/blob/master/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L1229-L1231,
which is not great.
This means that with sufficient task count overhead, a supervisor can OOM
as it fails to allocate more kernel threads for a supervisor. While the worker
threadpool's keep alive timeout is configured, critically it does not set
allowCoreThreadTimeOut(true), which actually enables the reaping of the idle
worker threads
(https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html).
This change enables that with a regression test.
---
.../util/common/concurrent/ScheduledExecutors.java | 10 ++++++--
.../common/concurrent/ScheduledExecutorsTest.java | 28 ++++++++++++++++++++++
2 files changed, 36 insertions(+), 2 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutors.java
b/processing/src/main/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutors.java
index 271ef85d130..bfd584f90d7 100644
---
a/processing/src/main/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutors.java
+++
b/processing/src/main/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutors.java
@@ -225,9 +225,14 @@ public class ScheduledExecutors
}
/**
- * Creates a new {@link ScheduledExecutorService} with a minimum number of
threads along with a
- * keep-alive time for idle non-core threads.
+ * Creates a new {@link ScheduledThreadPoolExecutor} sized to {@code
corePoolSize} whose idle threads
+ * are reclaimed after {@code keepAliveTimeInMillis}.
* <p>
+ * A {@link ScheduledThreadPoolExecutor} never grows beyond its core pool
size (its work queue is
+ * unbounded), so {@code corePoolSize} is really a ceiling. Without {@code
allowCoreThreadTimeOut(true)}
+ * the keep-alive would never apply to those core threads and they would
live for the lifetime of the
+ * pool even while idle. Enabling core-thread timeout lets the pool shrink
back toward zero when idle,
+ * so {@code corePoolSize} behaves as a peak-concurrency ceiling rather than
a permanent allocation.
*/
public static ScheduledThreadPoolExecutor fixedWithKeepAliveTime(
int corePoolSize,
@@ -240,6 +245,7 @@ public class ScheduledExecutors
Execs.makeThreadFactory(nameFormat)
);
scheduledExecutor.setKeepAliveTime(keepAliveTimeInMillis,
TimeUnit.MILLISECONDS);
+ scheduledExecutor.allowCoreThreadTimeOut(true);
return scheduledExecutor;
}
}
diff --git
a/processing/src/test/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutorsTest.java
b/processing/src/test/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutorsTest.java
index df046c1c0ad..8fe6eccbb4b 100644
---
a/processing/src/test/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutorsTest.java
+++
b/processing/src/test/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutorsTest.java
@@ -27,6 +27,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -333,4 +334,31 @@ public class ScheduledExecutorsTest
Assertions.assertTrue(completed, "Should continue executing after
exception");
Assertions.assertEquals(3, executionCount.get(), "Should have exactly 3
executions");
}
+
+ @Test
+ public void testFixedWithKeepAliveTimeReclaimsIdleThreads() throws Exception
+ {
+ final ScheduledThreadPoolExecutor exec =
ScheduledExecutors.fixedWithKeepAliveTime(8, "testKeepAlive-%d", 100);
+ try {
+ // Core threads must be eligible for timeout, otherwise a large
corePoolSize would pin that many
+ // threads for the lifetime of the pool even when idle.
+ Assertions.assertTrue(exec.allowsCoreThreadTimeOut(), "Core threads
should be allowed to time out");
+
+ final CountDownLatch latch = new CountDownLatch(4);
+ for (int i = 0; i < 4; i++) {
+ exec.submit(latch::countDown);
+ }
+ Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS), "Submitted tasks
should run");
+
+ // Once idle past the keep-alive, all workers should be reclaimed and
the pool shrink back to zero.
+ final long deadline = System.currentTimeMillis() + 5000;
+ while (exec.getPoolSize() > 0 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(50);
+ }
+ Assertions.assertEquals(0, exec.getPoolSize(), "Idle core threads should
be reclaimed");
+ }
+ finally {
+ exec.shutdownNow();
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
