This is an automated email from the ASF dual-hosted git repository.
FrankChen021 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 317feddae9a fix: Prevent a lost wakeup in SegmentLoadDropHandler from
stalling segment loading (#20608)
317feddae9a is described below
commit 317feddae9a73be7bb4b52cb54a7fa478c53011d
Author: Yahya Kisana <[email protected]>
AuthorDate: Fri Oct 30 18:10:13 2026 -0600
fix: Prevent a lost wakeup in SegmentLoadDropHandler from stalling segment
loading (#20608)
SegmentLoadDropHandler.processBatch() checked whether any request in the
batch was done and only afterwards registered the batch future in
waitingFutures. If a load finished in between, its callback ran
resolveWaitingFutures() before the future was registered, and the future
then waited for some other request on the server to finish. The
Coordinator's HttpLoadQueuePeon keeps one batch in flight per server, so
nothing else could finish, and the response came back only after the
async timeout (hostTimeout, 5 minutes by default). Until then the
server loaded no other segments.
Re-check the statuses while holding the waitingFutures lock, and register
the future only if no request has finished yet; otherwise resolve it
right away.
Fixes #20603
---
.../coordination/SegmentLoadDropHandler.java | 18 +++++--
.../coordination/SegmentLoadDropHandlerTest.java | 61 ++++++++++++++++++++++
2 files changed, 75 insertions(+), 4 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
b/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
index 50590bd275b..12c4e1cf5ed 100644
---
a/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
+++
b/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
@@ -308,14 +308,24 @@ public class SegmentLoadDropHandler
CustomSettableFuture future = new CustomSettableFuture(waitingFutures,
statuses);
- if (isAnyRequestDone) {
- future.resolve();
- } else {
+ if (!isAnyRequestDone) {
synchronized (waitingFutures) {
- waitingFutures.add(future);
+ // A request may have finished after its status was checked above, in
which case its callback has already
+ // called resolveWaitingFutures() and would never resolve this future.
Check again while holding the lock that
+ // resolveWaitingFutures() uses, so that the future is registered only
if no request has finished yet.
+ isAnyRequestDone = statuses.values().stream().anyMatch(
+ status -> status.get().getState() !=
SegmentChangeStatus.State.PENDING
+ );
+ if (!isAnyRequestDone) {
+ waitingFutures.add(future);
+ }
}
}
+ if (isAnyRequestDone) {
+ future.resolve();
+ }
+
return future;
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
index 13ae7ee19d3..6a9ea2a5952 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
@@ -49,6 +49,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
@@ -445,6 +446,66 @@ public class SegmentLoadDropHandlerTest
}
+ /**
+ * A load finishes, and its callback resolves the waiting futures, after
{@code processBatch()} has seen it as
+ * pending but before the future for the batch has been registered. The
future must still be resolved right away,
+ * rather than waiting for another request to finish or for the HTTP request
to time out.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS, threadMode =
ThreadMode.SEPARATE_THREAD)
+ public void
testProcessBatchResolvesWhenLoadFinishesBeforeFutureIsRegistered() throws
Exception
+ {
+ final SegmentManager segmentManager = Mockito.mock(SegmentManager.class);
+ Mockito.when(segmentManager.loadSegment(ArgumentMatchers.any()))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // Holds back each submitted load until the next one is submitted, and
then runs it on the submitting thread.
+ // The first load thus finishes after processBatch() has checked its
status, and the second load never finishes.
+ final List<Runnable> heldLoads = new ArrayList<>();
+ final ScheduledThreadPoolExecutor normalLoadExec =
+ new ScheduledThreadPoolExecutor(1,
Execs.makeThreadFactory("LoadDropHandlerTest-[%d]"))
+ {
+ @Override
+ public Future<?> submit(Runnable task)
+ {
+ if (!heldLoads.isEmpty()) {
+ heldLoads.remove(0).run();
+ }
+ heldLoads.add(task);
+ return null;
+ }
+ };
+
+ final SegmentLoadDropHandler handler = new SegmentLoadDropHandler(
+ segmentLoaderConfig,
+ segmentAnnouncer,
+ segmentManager,
+ normalLoadExec,
+ (ThreadPoolExecutor) scheduledExecutorFactory.create(5,
"TurboSegmentLoadDropHandlerTest-[%d]")
+ );
+
+ final DataSegment segment1 = makeSegment("batchtest1", "1",
Intervals.of("P1d/2011-04-01"));
+ final DataSegment segment2 = makeSegment("batchtest2", "1",
Intervals.of("P1d/2011-04-01"));
+ final List<DataSegmentChangeRequest> batch = ImmutableList.of(
+ new SegmentChangeRequestLoad(segment1),
+ new SegmentChangeRequestLoad(segment2)
+ );
+
+ final ListenableFuture<List<DataSegmentChangeResponse>> future =
+ handler.processBatch(batch, SegmentLoadingMode.NORMAL);
+ Assertions.assertEquals(1, heldLoads.size(), "Loads must go through the
overridden submit() to force the race");
+ Assertions.assertTrue(future.isDone());
+
+ final Map<DataSegmentChangeRequest, SegmentChangeStatus> expectedStatusMap
= Map.of(
+ batch.get(0), SegmentChangeStatus.success(SegmentLoadingMode.NORMAL),
+ batch.get(1), SegmentChangeStatus.pending(SegmentLoadingMode.NORMAL)
+ );
+ for (DataSegmentChangeResponse requestAndStatus : future.get()) {
+
Assertions.assertEquals(expectedStatusMap.get(requestAndStatus.getRequest()),
requestAndStatus.getStatus());
+ }
+ Assertions.assertEquals(ImmutableList.of(segment1),
segmentAnnouncer.getObservedSegments());
+ }
+
private SegmentLoadDropHandler initSegmentLoadDropHandler(SegmentManager
segmentManager)
{
return initSegmentLoadDropHandler(segmentLoaderConfig, segmentManager);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
