Skip to content

[Test] Stabilize Pipe concurrency tests #19143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[Test] Stabilize Pipe concurrency tests #19143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1dd10cf
[Test] Stabilize Pipe concurrency tests
Caideyipi Nov 5, 2026
1402474
Merge origin/master into Pipe concurrency test branch
Caideyipi Nov 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions ...datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -389,7 +389,7 @@ public void testSinkCriticalExceptionIsPropagatedOnlyWithinItsPipe() throws Exce
taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(failedTaskMeta, exception);

Awaitility.await()
.atMost(5, TimeUnit.SECONDS)
.atMost(30, TimeUnit.SECONDS)
.until(
() ->
PipeStatus.STOPPED.equals(failedPipeMeta.getRuntimeMeta().getStatus().get())
Expand Down Expand Up @@ -462,7 +462,7 @@ public void testDetachedTaskMetaMustIdentifyOnePipeUniquely() throws Exception {
taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(
detachedTaskMeta, new PipeRuntimeCriticalException("ambiguous task meta", 1L));

Awaitility.await().atMost(2, TimeUnit.SECONDS).until(detachedTaskMeta::hasExceptionMessages);
Awaitility.await().atMost(30, TimeUnit.SECONDS).until(detachedTaskMeta::hasExceptionMessages);
pipeMetaKeeper.acquireWriteLock();
pipeMetaKeeper.releaseWriteLock();

Expand Down Expand Up @@ -499,7 +499,7 @@ public void testDetachedTaskMetaIsRecordedOnIdentifiedPipe() throws Exception {
"detachedPipe", 2L, detachedTaskMeta, exception);

Awaitility.await()
.atMost(5, TimeUnit.SECONDS)
.atMost(30, TimeUnit.SECONDS)
.until(
() ->
PipeStatus.STOPPED.equals(pipeMeta.getRuntimeMeta().getStatus().get())
Expand Down Expand Up @@ -533,7 +533,7 @@ public void testCriticalExceptionWithNullTaskMetaStopsIdentifiedPipe() throws Ex
"nullTaskMetaPipe", 3L, null, exception);

Awaitility.await()
.atMost(5, TimeUnit.SECONDS)
.atMost(30, TimeUnit.SECONDS)
.until(
() ->
PipeStatus.STOPPED.equals(failedPipeMeta.getRuntimeMeta().getStatus().get())
Expand Down
13 changes: 7 additions & 6 deletions ...tanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ protected boolean executeOnce() {
if (current == 1) {
firstCallEntered.countDown();
try {
firstCallReleased.await(5, TimeUnit.SECONDS);
firstCallReleased.await();
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand All @@ -171,13 +171,14 @@ protected boolean executeOnce() {
};

executor.register(blockedSubtask);
executor.start(blockedSubtask.getTaskID());
Assert.assertTrue(firstCallEntered.await(5, TimeUnit.SECONDS));

executor.stop(blockedSubtask.getTaskID());
executor.start(blockedSubtask.getTaskID());

try {
executor.start(blockedSubtask.getTaskID());
Assert.assertTrue(firstCallEntered.await(30, TimeUnit.SECONDS));

executor.stop(blockedSubtask.getTaskID());
executor.start(blockedSubtask.getTaskID());

Assert.assertFalse(
"The subtask must not be executed in parallel with itself",
secondCallEntered.await(1, TimeUnit.SECONDS));
Expand Down
46 changes: 29 additions & 17 deletions ...he/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws Exceptio
}
}

@Test(timeout = 60000)
@Test(timeout = 120000)
public void testCloseCancelsReservationEnqueuedByAWorkerInvocation() throws Exception {
final CommonConfig config = CommonDescriptor.getInstance().getConfig();
final PipeMemoryManager manager = PipeDataNodeResourceManager.memory();
Expand Down Expand Up @@ -226,7 +226,7 @@ public void testCloseCancelsReservationEnqueuedByAWorkerInvocation() throws Exce
eventToClose.close();
});
await()
.atMost(5, TimeUnit.SECONDS)
.atMost(30, TimeUnit.SECONDS)
.until(
() ->
closingThread.get() != null
Expand All @@ -241,27 +241,39 @@ public void testCloseCancelsReservationEnqueuedByAWorkerInvocation() throws Exce
Assert.assertEquals(1, manager.getWaitingTsFileParserCount());
Assert.assertTrue(outputQueue.isEmpty());
}
closing.get(5, TimeUnit.SECONDS);
closing.get(30, TimeUnit.SECONDS);
Assert.assertFalse(event.isWaitingForTsFileParserMemory());
Assert.assertEquals(0, manager.getWaitingTsFileParserCount());
Assert.assertEquals(1, manager.getReservedTsFileParserCount());
} finally {
executor.shutdownNow();
Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
if (subtask != null) {
subtask.disallowSubmittingSelf();
subtask.close();
}
if (event != null) {
event.close();
}
if (blockerReserved) {
manager.releaseTsFileParserMemory("blocker", 0, "0");
try {
Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
} finally {
try {
if (subtask != null) {
subtask.disallowSubmittingSelf();
subtask.close();
}
} finally {
try {
if (event != null) {
event.close();
}
} finally {
try {
if (blockerReserved) {
manager.releaseTsFileParserMemory("blocker", 0, "0");
}
} finally {
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
FileUtils.deleteFileOrDirectory(tempDir);
}
}
}
}
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
FileUtils.deleteFileOrDirectory(tempDir);
}
}

Expand Down
84 changes: 51 additions & 33 deletions ...org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ public void testRequestsOnOneWorkerDoNotStrandOtherWorkersParserSlots() throws E
}
}

@Test(timeout = 30000)
@Test(timeout = 120000)
public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Exception {
final CommonConfig config = CommonDescriptor.getInstance().getConfig();
final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory();
Expand Down Expand Up @@ -234,8 +234,8 @@ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Excepti
final AtomicReference<Thread> workerThread = workerThreads.get(i);
// Consume the scheduling notification before all workers become idle on admission.
worker.sleepIfNecessary(true);
// Make a missed wake-up observable without relying on a tight performance assertion.
worker.setSleepingTimeInMilliSecond(60000L);
// Keep polling beyond the whole test timeout so it cannot hide a missed wake-up.
worker.setSleepingTimeInMilliSecond(180000L);
futures.add(
executor.submit(
() -> {
Expand All @@ -246,7 +246,7 @@ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Excepti
}));
}
await()
.atMost(5, TimeUnit.SECONDS)
.atMost(30, TimeUnit.SECONDS)
.until(
() ->
workerThreads.stream()
Expand All @@ -260,25 +260,34 @@ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Excepti
}
Assert.assertTrue(
"Admission notifications must wake every queue head",
processed.await(5, TimeUnit.SECONDS));
processed.await(30, TimeUnit.SECONDS));
Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount());
Assert.assertEquals(0, memoryManager.getReservedTsFileParserCount());
} finally {
futures.forEach(future -> future.cancel(true));
executor.shutdownNow();
Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
for (int i = 0; i < requests.size(); i++) {
memoryManager.cancelTsFileParserMemoryReservation(
"waitingPipe" + i, 0, "0", requests.get(i));
}
for (int i = 0; i < reservedBlockers.length; i++) {
if (reservedBlockers[i]) {
memoryManager.releaseTsFileParserMemory("blocker" + i, 0, "0");
try {
Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
} finally {
try {
for (int i = 0; i < requests.size(); i++) {
memoryManager.cancelTsFileParserMemoryReservation(
"waitingPipe" + i, 0, "0", requests.get(i));
}
} finally {
try {
for (int i = 0; i < reservedBlockers.length; i++) {
if (reservedBlockers[i]) {
memoryManager.releaseTsFileParserMemory("blocker" + i, 0, "0");
}
}
} finally {
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
}
}
}
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
}
}

Expand Down Expand Up @@ -334,7 +343,7 @@ public void testWaitingParserIsSkippedUntilAdmissionNotification() throws Except
}
}

@Test(timeout = 10000)
@Test(timeout = 90000)
public void testNotificationBeforeIdleWaitIsNotLost() throws Exception {
final PipeProcessorSubtaskWorker worker = newWorker(new LinkedHashSet<>());
worker.setSleepingTimeInMilliSecond(60000L);
Expand All @@ -352,14 +361,14 @@ public void testNotificationBeforeIdleWaitIsNotLost() throws Exception {
final ExecutorService executor = Executors.newSingleThreadExecutor();
try {
wakeUpCallback.get().run();
executor.submit(() -> worker.sleepIfNecessary(true)).get(3, TimeUnit.SECONDS);
executor.submit(() -> worker.sleepIfNecessary(true)).get(30, TimeUnit.SECONDS);
} finally {
executor.shutdownNow();
Assert.assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS));
Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
}
}

@Test(timeout = 15000)
@Test(timeout = 150000)
public void testBusyFirstWorkerDoesNotSerializeOtherEligibleWorker() throws Exception {
final CommonConfig config = CommonDescriptor.getInstance().getConfig();
final PipeMemoryManager manager = PipeDataNodeResourceManager.memory();
Expand Down Expand Up @@ -393,7 +402,7 @@ public void testBusyFirstWorkerDoesNotSerializeOtherEligibleWorker() throws Exce
ignored -> {
if (blockFirstWorker.get()) {
firstWorkerBusy.countDown();
Assert.assertTrue(unblockFirstWorker.await(5, TimeUnit.SECONDS));
unblockFirstWorker.await();
}
return false;
});
Expand Down Expand Up @@ -447,7 +456,7 @@ public void testBusyFirstWorkerDoesNotSerializeOtherEligibleWorker() throws Exce
Assert.assertTrue(secondWorker.runSubtasks());
blockFirstWorker.set(true);
final Future<?> busyRound = executor.submit(firstWorker::runSubtasks);
Assert.assertTrue(firstWorkerBusy.await(3, TimeUnit.SECONDS));
Assert.assertTrue(firstWorkerBusy.await(30, TimeUnit.SECONDS));
for (int i = 0; i < reservedBlockers.length; i++) {
manager.releaseTsFileParserMemory("blocker" + i, 0, "0");
reservedBlockers[i] = false;
Expand All @@ -458,27 +467,36 @@ public void testBusyFirstWorkerDoesNotSerializeOtherEligibleWorker() throws Exce
secondWorker.sleepIfNecessary(secondWorker.runSubtasks());
}
});
Assert.assertTrue(secondAdmitted.await(3, TimeUnit.SECONDS));
Assert.assertTrue(secondAdmitted.await(30, TimeUnit.SECONDS));
verify(firstSubtask, times(1)).call();
Assert.assertFalse(busyRound.isDone());
unblockFirstWorker.countDown();
busyRound.get(3, TimeUnit.SECONDS);
busyRound.get(30, TimeUnit.SECONDS);
Assert.assertEquals(0, manager.getWaitingTsFileParserCount());
Assert.assertEquals(0, manager.getReservedTsFileParserCount());
} finally {
unblockFirstWorker.countDown();
executor.shutdownNow();
Assert.assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS));
manager.cancelTsFileParserMemoryReservation("first", 0, "0", first);
manager.cancelTsFileParserMemoryReservation("second", 0, "0", second);
for (int i = 0; i < reservedBlockers.length; i++) {
if (reservedBlockers[i]) {
manager.releaseTsFileParserMemory("blocker" + i, 0, "0");
try {
Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
} finally {
try {
manager.cancelTsFileParserMemoryReservation("first", 0, "0", first);
manager.cancelTsFileParserMemoryReservation("second", 0, "0", second);
} finally {
try {
for (int i = 0; i < reservedBlockers.length; i++) {
if (reservedBlockers[i]) {
manager.releaseTsFileParserMemory("blocker" + i, 0, "0");
}
}
} finally {
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
}
}
}
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
}
}

Expand Down
Loading
Loading