This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 4fa3bac0093 [BUGFIX] Fix interrupted scaling-pool submissions (#19222)
4fa3bac0093 is described below
commit 4fa3bac0093b6940a8099c34c83862e8c0c88e34
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Aug 11 21:20:23 2026 -0700
[BUGFIX] Fix interrupted scaling-pool submissions (#19222)
---
.../common/utils/ScalingThreadPoolExecutor.java | 27 ++++++--
.../utils/ScalingThreadPoolExecutorTest.java | 50 +++++++++++++++
.../invertedindex/LuceneMutableTextIndexTest.java | 73 +++++++++++++++++++---
3 files changed, 137 insertions(+), 13 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutor.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutor.java
index 1f38af2888f..09457e9473b 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutor.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutor.java
@@ -72,19 +72,27 @@ public class ScalingThreadPoolExecutor extends
ThreadPoolExecutor {
public static ExecutorService newScalingThreadPool(int min, int max, long
keepAliveTime) {
ScalingQueue<Runnable> queue = new ScalingQueue<>();
ThreadPoolExecutor executor = new ScalingThreadPoolExecutor(min, max,
keepAliveTime, TimeUnit.MILLISECONDS, queue);
- executor.setRejectedExecutionHandler(new ForceQueuePolicy());
+ executor.setRejectedExecutionHandler(new ForceQueuePolicy(queue));
return executor;
}
/// Used to handle queue rejections. The policy ensures we still queue the
Runnable, and the rejection ensures the
/// pool will be expanded if necessary
static class ForceQueuePolicy implements RejectedExecutionHandler {
+ private final ScalingQueue<Runnable> _queue;
+
+ ForceQueuePolicy(ScalingQueue<Runnable> queue) {
+ _queue = queue;
+ }
+
+ @Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
- try {
- executor.getQueue().put(r);
- } catch (InterruptedException e) {
- // should never happen since we never wait
- throw new RejectedExecutionException(e);
+ if (executor.isShutdown() || !_queue.forceOffer(r)) {
+ throw new RejectedExecutionException("Task " + r + " rejected from " +
executor);
+ }
+ // Match ThreadPoolExecutor's queueing recheck: do not leave a task
stranded if shutdown raced the offer.
+ if (executor.isShutdown() && executor.remove(r)) {
+ throw new RejectedExecutionException("Task " + r + " rejected from " +
executor);
}
}
}
@@ -132,5 +140,12 @@ public class ScalingThreadPoolExecutor extends
ThreadPoolExecutor {
public boolean offer(E e) {
return _currentIdleThreadCount.get() > 0 && super.offer(e);
}
+
+ /// Unconditionally offers an element after the executor has reached its
maximum pool size.
+ /// LinkedBlockingQueue.offer is non-interruptible, which allows an
already-interrupted caller
+ /// to queue work while preserving its interrupt status.
+ boolean forceOffer(E e) {
+ return super.offer(e);
+ }
}
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutorTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutorTest.java
index 7f0e892a468..0942ecb17dd 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutorTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ScalingThreadPoolExecutorTest.java
@@ -18,11 +18,16 @@
*/
package org.apache.pinot.common.utils;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import org.apache.pinot.util.TestUtils;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
@@ -72,6 +77,51 @@ public class ScalingThreadPoolExecutorTest {
"Timed out waiting for thread pool to scale up");
}
+ @Test
+ public void testInterruptedSubmissionIsQueued() throws Exception {
+ ThreadPoolExecutor executorService =
+ (ThreadPoolExecutor) ScalingThreadPoolExecutor.newScalingThreadPool(0,
1, 500);
+ CountDownLatch taskStarted = new CountDownLatch(1);
+ CountDownLatch releaseTask = new CountDownLatch(1);
+ try {
+ Future<?> activeTask = executorService.submit(() -> {
+ taskStarted.countDown();
+ try {
+ releaseTask.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(taskStarted.await(10, TimeUnit.SECONDS), "Timed out waiting
for the active task to start");
+
+ Future<?> queuedTask;
+ try {
+ Thread.currentThread().interrupt();
+ queuedTask = executorService.submit(() -> { });
+ assertTrue(Thread.currentThread().isInterrupted(), "Submission should
preserve the caller's interrupt status");
+ } finally {
+ Thread.interrupted();
+ releaseTask.countDown();
+ }
+
+ activeTask.get(10, TimeUnit.SECONDS);
+ queuedTask.get(10, TimeUnit.SECONDS);
+ } finally {
+ Thread.interrupted();
+ releaseTask.countDown();
+ executorService.shutdownNow();
+ assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS),
"Executor did not terminate");
+ }
+ }
+
+ @Test
+ public void testSubmissionAfterShutdownIsRejected() {
+ ThreadPoolExecutor executorService =
+ (ThreadPoolExecutor) ScalingThreadPoolExecutor.newScalingThreadPool(0,
1, 500);
+ executorService.shutdown();
+ assertThrows(RejectedExecutionException.class, () ->
executorService.submit(() -> { }));
+ }
+
private Runnable getSleepingRunnable() {
return () -> {
try {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
index f6afbf9a37c..8d71f6a21f0 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
@@ -21,6 +21,7 @@ package
org.apache.pinot.segment.local.realtime.impl.invertedindex;
import java.io.File;
import java.io.IOException;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -46,6 +47,8 @@ import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
public class LuceneMutableTextIndexTest {
@@ -317,18 +320,74 @@ public class LuceneMutableTextIndexTest {
assertEquals(_realtimeLuceneTextIndex.getDocIds("invalid"),
ImmutableRoaringBitmap.bitmapOf());
}
- @Test(expectedExceptions = ExecutionException.class,
- expectedExceptionsMessageRegExp = ".*TEXT_MATCH query interrupted while
querying the consuming segment.*")
+ @Test
public void testQueryCancellationIsSuccessful()
- throws InterruptedException, ExecutionException {
+ throws Exception {
configureIndex(null, null, null, null);
+ CountDownLatch searcherTaskStarted = new CountDownLatch(1);
+ CountDownLatch releaseSearcherTask = new CountDownLatch(1);
+ Future<?> searcherTask = SEARCHER_POOL.getExecutorService().submit(() -> {
+ searcherTaskStarted.countDown();
+ awaitUninterruptibly(releaseSearcherTask);
+ });
+
// Avoid early finalization by not using Executors.newSingleThreadExecutor
(java <= 20, JDK-8145304)
ExecutorService baseExecutor = Executors.newFixedThreadPool(1);
// Wrap with contextAwareExecutorService to propagate QueryThreadContext
to child threads
ExecutorService executor =
QueryThreadContext.contextAwareExecutorService(baseExecutor);
- Future<MutableRoaringBitmap> res = executor.submit(() ->
_realtimeLuceneTextIndex.getDocIds("/.*read.*/"));
- // Shutdown the base executor to trigger interrupt on the worker thread
- baseExecutor.shutdownNow();
- res.get();
+ CountDownLatch queryTaskStarted = new CountDownLatch(1);
+ CountDownLatch runQuery = new CountDownLatch(1);
+
+ try {
+ assertTrue(searcherTaskStarted.await(10, TimeUnit.SECONDS), "Timed out
waiting for searcher task to start");
+ Future<MutableRoaringBitmap> result = executor.submit(() -> {
+ queryTaskStarted.countDown();
+ awaitUninterruptibly(runQuery);
+ return _realtimeLuceneTextIndex.getDocIds("/.*read.*/");
+ });
+ assertTrue(queryTaskStarted.await(10, TimeUnit.SECONDS), "Timed out
waiting for query task to start");
+
+ // Interrupt the query worker before it submits to the saturated
searcher pool, reproducing the CI race.
+ baseExecutor.shutdownNow();
+ runQuery.countDown();
+
+ ExecutionException exception =
+ expectThrows(ExecutionException.class, () -> result.get(10,
TimeUnit.SECONDS));
+ Throwable cause = exception.getCause();
+ assertTrue(cause instanceof RuntimeException, "Expected RuntimeException
but got: " + cause);
+ assertTrue(cause.getMessage().contains("TEXT_MATCH query interrupted
while querying the consuming segment"),
+ "Unexpected exception: " + cause);
+ } finally {
+ runQuery.countDown();
+ baseExecutor.shutdownNow();
+ boolean queryExecutorTerminated = baseExecutor.awaitTermination(10,
TimeUnit.SECONDS);
+
+ // Queue a marker behind the nested Lucene search, then release the
single searcher thread. Waiting for the
+ // marker guarantees the search completed before tearDownMethod closes
the index.
+ Future<?> searcherDrain;
+ try {
+ searcherDrain = SEARCHER_POOL.getExecutorService().submit(() -> { });
+ } finally {
+ releaseSearcherTask.countDown();
+ }
+ searcherTask.get(10, TimeUnit.SECONDS);
+ searcherDrain.get(10, TimeUnit.SECONDS);
+ assertTrue(queryExecutorTerminated, "Query executor did not terminate");
+ }
+ }
+
+ private static void awaitUninterruptibly(CountDownLatch latch) {
+ boolean interrupted = false;
+ while (true) {
+ try {
+ latch.await();
+ break;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
