diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
index f86b185d07fb8..0d0c0e335ddc9 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
@@ -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())
@@ -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();
 
@@ -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())
@@ -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())
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java
index bb24ac2a7792c..6c673017d6aaf 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSinkSubtaskExecutorTest.java
@@ -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();
               }
@@ -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));
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
index a4a65a5327189..cf693271c24cb 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
@@ -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();
@@ -226,7 +226,7 @@ public void testCloseCancelsReservationEnqueuedByAWorkerInvocation() throws Exce
                   eventToClose.close();
                 });
         await()
-            .atMost(5, TimeUnit.SECONDS)
+            .atMost(30, TimeUnit.SECONDS)
             .until(
                 () ->
                     closingThread.get() != null
@@ -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);
     }
   }
 
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
index 5fd36c4af40a0..190684b26e644 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
@@ -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();
@@ -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(
                 () -> {
@@ -246,7 +246,7 @@ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Excepti
                 }));
       }
       await()
-          .atMost(5, TimeUnit.SECONDS)
+          .atMost(30, TimeUnit.SECONDS)
           .until(
               () ->
                   workerThreads.stream()
@@ -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);
     }
   }
 
@@ -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);
@@ -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();
@@ -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;
               });
@@ -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;
@@ -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);
     }
   }
 
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
index eac860f8763a7..f15cfa4fad941 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
@@ -45,9 +45,11 @@
 import org.mockito.Mockito;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
@@ -116,31 +118,44 @@ public void testExternalSinkDiscardEventsOfPipeNotBlockedByConnectionRetry() thr
 
     final Thread failureThread =
         new Thread(() -> subtask.onFailure(new PipeConnectionException("connection broken")));
-    failureThread.start();
-    Assert.assertTrue(handshakeEntered.await(5, TimeUnit.SECONDS));
-
-    final CountDownLatch discardReturned = new CountDownLatch(1);
-    final Thread discardThread =
-        new Thread(
-            () -> {
-              try {
-                subtask.discardEventsOfPipe(new CommitterKey("pipe", 1L, 1, -1));
-              } finally {
-                discardReturned.countDown();
-              }
-            });
-    discardThread.start();
+    failureThread.setDaemon(true);
+    Thread discardThread = null;
 
     try {
-      Assert.assertTrue(discardReturned.await(5, TimeUnit.SECONDS));
+      failureThread.start();
+      Assert.assertTrue(handshakeEntered.await(30, TimeUnit.SECONDS));
+
+      final CountDownLatch discardReturned = new CountDownLatch(1);
+      discardThread =
+          new Thread(
+              () -> {
+                try {
+                  subtask.discardEventsOfPipe(new CommitterKey("pipe", 1L, 1, -1));
+                } finally {
+                  discardReturned.countDown();
+                }
+              });
+      discardThread.setDaemon(true);
+      discardThread.start();
+
+      Assert.assertTrue(discardReturned.await(30, TimeUnit.SECONDS));
       Assert.assertEquals(1L, discardEntered.getCount());
       Assert.assertFalse(discardDuringHandshake.get());
     } finally {
       releaseHandshake.countDown();
-      failureThread.join(5000);
-      Assert.assertTrue(discardEntered.await(1, TimeUnit.SECONDS));
-      Assert.assertFalse(discardDuringHandshake.get());
-      subtask.close();
+      try {
+        failureThread.join(30_000);
+        Assert.assertFalse(failureThread.isAlive());
+        if (discardThread != null) {
+          discardThread.join(30_000);
+          Assert.assertFalse(discardThread.isAlive());
+          // Closing the subtask also drains pending discards. Check the handshake path first.
+          Assert.assertTrue(discardEntered.await(30, TimeUnit.SECONDS));
+          Assert.assertFalse(discardDuringHandshake.get());
+        }
+      } finally {
+        subtask.close();
+      }
     }
   }
 
@@ -174,9 +189,7 @@ public void testBuiltinSinkDiscardEventsOfPipeWaitsForConnectionRetry() throws E
 
     final Thread failureThread =
         new Thread(() -> subtask.onFailure(new PipeConnectionException("connection broken")));
-    failureThread.start();
-    Assert.assertTrue(handshakeEntered.await(5, TimeUnit.SECONDS));
-
+    failureThread.setDaemon(true);
     final CountDownLatch discardReturned = new CountDownLatch(1);
     final Thread discardThread =
         new Thread(
@@ -187,22 +200,40 @@ public void testBuiltinSinkDiscardEventsOfPipeWaitsForConnectionRetry() throws E
                 discardReturned.countDown();
               }
             });
-    discardThread.start();
+    discardThread.setDaemon(true);
 
     try {
-      Assert.assertFalse(discardReturned.await(100, TimeUnit.MILLISECONDS));
+      failureThread.start();
+      Assert.assertTrue(handshakeEntered.await(30, TimeUnit.SECONDS));
+
+      discardThread.start();
+      await()
+          .atMost(30, TimeUnit.SECONDS)
+          .until(
+              () ->
+                  discardThread.getState() == Thread.State.WAITING
+                      || discardReturned.getCount() == 0);
+      Assert.assertEquals(1L, discardReturned.getCount());
       Assert.assertEquals(1L, discardEntered.getCount());
       Assert.assertFalse(discardDuringHandshake.get());
+
+      releaseHandshake.countDown();
+      discardThread.join(30_000);
+      failureThread.join(30_000);
+      Assert.assertFalse(discardThread.isAlive());
+      Assert.assertFalse(failureThread.isAlive());
+      // Closing the subtask also drains pending discards. Check normal completion first.
+      Assert.assertTrue(discardEntered.await(30, TimeUnit.SECONDS));
+      Assert.assertFalse(discardDuringHandshake.get());
     } finally {
       releaseHandshake.countDown();
-      discardThread.join(5000);
-      failureThread.join(5000);
-      subtask.close();
+      try {
+        discardThread.join(30_000);
+        failureThread.join(30_000);
+      } finally {
+        subtask.close();
+      }
     }
-
-    Assert.assertFalse(discardThread.isAlive());
-    Assert.assertTrue(discardEntered.await(1, TimeUnit.SECONDS));
-    Assert.assertFalse(discardDuringHandshake.get());
   }
 
   @Test
@@ -237,24 +268,45 @@ public void testExternalSinkCloseNotConcurrentWithConnectionRetry() throws Excep
 
     final Thread failureThread =
         new Thread(() -> subtask.onFailure(new PipeConnectionException("connection broken")));
-    failureThread.start();
+    failureThread.setDaemon(true);
+    final FutureTask<Long> closeTask =
+        new FutureTask<>(
+            () -> {
+              final long startTime = System.nanoTime();
+              subtask.close();
+              return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
+            });
+    final Thread closeThread = new Thread(closeTask);
+    closeThread.setDaemon(true);
+    boolean closeStarted = false;
 
     try {
-      Assert.assertTrue(handshakeEntered.await(5, TimeUnit.SECONDS));
+      failureThread.start();
+      Assert.assertTrue(handshakeEntered.await(30, TimeUnit.SECONDS));
 
-      final long startTime = System.currentTimeMillis();
-      subtask.close();
+      closeThread.start();
+      closeStarted = true;
 
-      Assert.assertTrue(System.currentTimeMillis() - startTime < 1000);
+      Assert.assertTrue(closeTask.get(30, TimeUnit.SECONDS) < 1000);
       Assert.assertFalse(closeEntered.await(100, TimeUnit.MILLISECONDS));
       Assert.assertFalse(closeDuringHandshake.get());
     } finally {
       releaseHandshake.countDown();
-      Assert.assertTrue(closeEntered.await(1, TimeUnit.SECONDS));
-      failureThread.join(5000);
-      Assert.assertFalse(closeDuringHandshake.get());
-      CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
+      try {
+        if (!closeStarted) {
+          subtask.close();
+        }
+        closeThread.join(30_000);
+        failureThread.join(30_000);
+        Assert.assertFalse(closeThread.isAlive());
+        Assert.assertTrue(closeEntered.await(30, TimeUnit.SECONDS));
+        Assert.assertFalse(closeDuringHandshake.get());
+      } finally {
+        CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
+      }
     }
+
+    Assert.assertFalse(failureThread.isAlive());
   }
 
   @Test
@@ -290,9 +342,10 @@ public void testExternalSinkCloseDoesNotWaitForeverForConnectorClose() throws Ex
     try {
       final long startTime = System.currentTimeMillis();
       subtask.close();
+      final long closeDuration = System.currentTimeMillis() - startTime;
 
-      Assert.assertTrue(closeEntered.await(1, TimeUnit.SECONDS));
-      Assert.assertTrue(System.currentTimeMillis() - startTime < 1000);
+      Assert.assertTrue(closeDuration < 1000);
+      Assert.assertTrue(closeEntered.await(30, TimeUnit.SECONDS));
     } finally {
       releaseClose.countDown();
       CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
@@ -313,7 +366,7 @@ public void testBuiltinSinkCloseWaitsForConnectorClose() throws Exception {
     doAnswer(
             invocation -> {
               closeEntered.countDown();
-              releaseClose.await(5, TimeUnit.SECONDS);
+              releaseClose.await();
               return null;
             })
         .when(connector)
@@ -340,15 +393,19 @@ public void testBuiltinSinkCloseWaitsForConnectorClose() throws Exception {
                 closeReturned.countDown();
               }
             });
+    closeThread.setDaemon(true);
 
     try {
       closeThread.start();
-      Assert.assertTrue(closeEntered.await(5, TimeUnit.SECONDS));
+      Assert.assertTrue(closeEntered.await(30, TimeUnit.SECONDS));
       Assert.assertFalse(closeReturned.await(100, TimeUnit.MILLISECONDS));
     } finally {
       releaseClose.countDown();
-      closeThread.join(5000);
-      CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
+      try {
+        closeThread.join(30_000);
+      } finally {
+        CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
+      }
     }
 
     Assert.assertFalse(closeThread.isAlive());
@@ -623,7 +680,7 @@ public void handshake() throws Exception {
       handshaking = true;
       handshakeEntered.countDown();
       try {
-        releaseHandshake.await(5, TimeUnit.SECONDS);
+        releaseHandshake.await();
       } finally {
         handshaking = false;
       }
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java
index 04d7141096234..8718f3ec43896 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java
@@ -32,9 +32,10 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
+import static org.awaitility.Awaitility.await;
 import static org.mockito.Mockito.when;
 
 public class PipeDataRegionSinkMetricsTest {
@@ -52,7 +53,7 @@ public void testDeregisterWaitsForBind() throws Exception {
             invocation -> {
               if (blockBindOnce.compareAndSet(true, false)) {
                 bindEntered.countDown();
-                Assert.assertTrue(continueBind.await(5, TimeUnit.SECONDS));
+                continueBind.await();
               }
               return "pipe";
             });
@@ -71,28 +72,35 @@ public void testDeregisterWaitsForBind() throws Exception {
 
     final ExecutorService executor = Executors.newFixedThreadPool(2);
     final Future<?> bindFuture = executor.submit(() -> metrics.bindTo(metricService));
-    Future<?> deregisterFuture = null;
     try {
-      Assert.assertTrue(bindEntered.await(5, TimeUnit.SECONDS));
-      deregisterFuture = executor.submit(() -> metrics.deregister(taskId));
+      Assert.assertTrue(bindEntered.await(30, TimeUnit.SECONDS));
+      final AtomicReference<Thread> deregisterThread = new AtomicReference<>();
+      final Future<?> deregisterFuture =
+          executor.submit(
+              () -> {
+                deregisterThread.set(Thread.currentThread());
+                metrics.deregister(taskId);
+              });
       try {
-        try {
-          deregisterFuture.get(200, TimeUnit.MILLISECONDS);
-          Assert.fail("Deregistration must wait until metric binding completes");
-        } catch (final TimeoutException expected) {
-          // Expected: bindTo holds the lifecycle lock until all metrics are created.
-        }
+        await()
+            .atMost(30, TimeUnit.SECONDS)
+            .until(
+                () ->
+                    deregisterFuture.isDone()
+                        || (deregisterThread.get() != null
+                            && deregisterThread.get().getState() == Thread.State.BLOCKED));
+        Assert.assertFalse(
+            "Deregistration must wait until metric binding completes", deregisterFuture.isDone());
       } finally {
         continueBind.countDown();
       }
 
-      bindFuture.get(5, TimeUnit.SECONDS);
-      Assert.assertNotNull(deregisterFuture);
-      deregisterFuture.get(5, TimeUnit.SECONDS);
+      bindFuture.get(30, TimeUnit.SECONDS);
+      deregisterFuture.get(30, TimeUnit.SECONDS);
     } finally {
       continueBind.countDown();
       executor.shutdownNow();
-      Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+      Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
     }
   }
 }
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgentTest.java
index 0335e208fc3af..b3319609a5b9f 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgentTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgentTest.java
@@ -52,12 +52,12 @@
 
 public class IoTDBAirGapReceiverAgentTest {
 
-  @Test(timeout = 30_000)
+  @Test(timeout = 150_000)
   public void testUdpSocketIsDrainedWhileRequestHandlingIsBlocked() throws Exception {
     verifyUdpQueue(false);
   }
 
-  @Test(timeout = 30_000)
+  @Test(timeout = 150_000)
   public void testStoppingUdpReceiverReleasesQueuedPackets() throws Exception {
     verifyUdpQueue(true);
   }
@@ -97,12 +97,12 @@ void handleUdpPacket(final DatagramPacket packet) {
       try {
         final byte[] data = new byte[60 * 1024];
         sender.send(new DatagramPacket(data, data.length, receiver.getLocalSocketAddress()));
-        Assert.assertTrue(handlingStarted.await(5, TimeUnit.SECONDS));
+        Assert.assertTrue(handlingStarted.await(30, TimeUnit.SECONDS));
+        final long queueDeadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
         for (int i = 1; i <= 16; i++) {
           data[0] = (byte) i;
           sender.send(new DatagramPacket(data, data.length, receiver.getLocalSocketAddress()));
-          final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
-          while (queue.size() < i && System.nanoTime() < deadline) {
+          while (queue.size() < i && System.nanoTime() < queueDeadline) {
             Thread.sleep(1);
           }
           Assert.assertEquals(i, queue.size());
@@ -110,7 +110,7 @@ void handleUdpPacket(final DatagramPacket packet) {
         Assert.assertEquals(initialPermits - 17 * data.length, bytes.availablePermits());
         if (!stopWithPendingPackets) {
           resumeHandling.countDown();
-          Assert.assertTrue(handled.await(5, TimeUnit.SECONDS));
+          Assert.assertTrue(handled.await(30, TimeUnit.SECONDS));
           for (int i = 0; i < 17; i++) {
             Assert.assertEquals(i, order.get(i).intValue());
           }
@@ -118,7 +118,7 @@ void handleUdpPacket(final DatagramPacket packet) {
       } finally {
         agent.stop();
         resumeHandling.countDown();
-        Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+        Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
       }
       Assert.assertTrue(queue.isEmpty());
       Assert.assertEquals(initialPermits, bytes.availablePermits());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
index 1051fe8003093..28fc7a4e8628b 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
@@ -782,7 +782,8 @@ public void testConcurrentTsFilesFromMultiplePipesAreNotStarved() throws Excepti
 
                   boolean acquired = tryAcquireWithoutTracking(reservation);
                   enqueued.countDown();
-                  final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+                  // Allow time for all contenders to enqueue before the blocker is released.
+                  final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
                   while (!acquired && System.nanoTime() < deadline) {
                     reservation.key.await(
                         Math.max(
@@ -802,20 +803,20 @@ public void testConcurrentTsFilesFromMultiplePipesAreNotStarved() throws Excepti
                 }));
       }
 
-      Assert.assertTrue(ready.await(5, TimeUnit.SECONDS));
+      Assert.assertTrue(ready.await(30, TimeUnit.SECONDS));
       start.countDown();
-      Assert.assertTrue(enqueued.await(5, TimeUnit.SECONDS));
+      Assert.assertTrue(enqueued.await(30, TimeUnit.SECONDS));
       release(blocker);
 
       for (final Future<Boolean> future : futures) {
-        Assert.assertTrue(future.get(15, TimeUnit.SECONDS));
+        Assert.assertTrue(future.get(30, TimeUnit.SECONDS));
       }
       Assert.assertEquals(waitingTsFiles.size(), acquisitionOrder.size());
       Assert.assertEquals(3, new HashSet<>(acquisitionOrder.subList(0, 3)).size());
     } finally {
       release(blocker);
       executor.shutdownNow();
-      Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+      Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
     }
   }
 
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
index e5b016942a126..266610458cf56 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
@@ -424,7 +424,7 @@ public void testUdpAirGapSlicing() throws Exception {
         senderFinished.set(true);
       }
 
-      receiverThread.join(5000);
+      receiverThread.join(30_000);
       Assert.assertFalse(receiverThread.isAlive());
       Assert.assertNull(receiverException.get());
       Assert.assertTrue(receivedRequests.size() > 1);
@@ -487,7 +487,7 @@ public void testUdpAirGapRejectsOversizedAcknowledgement() throws Exception {
         receiverSocket.close();
       }
 
-      receiverThread.join(5000);
+      receiverThread.join(30_000);
       Assert.assertFalse(receiverThread.isAlive());
       Assert.assertNull(receiverException.get());
       Assert.assertTrue(oversizedAcknowledgementSent.get());
@@ -539,7 +539,7 @@ public void testUdpAirGapIgnoresStaleAcknowledgement() throws Exception {
         receiverSocket.close();
       }
 
-      receiverThread.join(5000);
+      receiverThread.join(30_000);
       Assert.assertFalse(receiverThread.isAlive());
       Assert.assertNull(receiverException.get());
       Assert.assertTrue(secondPacketReceived.get());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java
index ce208943a97d6..df634424b915b 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java
@@ -321,15 +321,15 @@ public void clearEventsReferenceCount() {
     final Thread closeThread = new Thread(sink::close, "pipe-sink-close");
     closeThread.setDaemon(true);
 
-    callbackThread.start();
-    Assert.assertTrue(callbackEntered.await(5, TimeUnit.SECONDS));
-    closeThread.start();
-    Assert.assertTrue(sink.closeEliminationStarted.await(5, TimeUnit.SECONDS));
-
     try {
+      callbackThread.start();
+      Assert.assertTrue(callbackEntered.await(30, TimeUnit.SECONDS));
+      closeThread.start();
+      Assert.assertTrue(sink.closeEliminationStarted.await(30, TimeUnit.SECONDS));
+
       allowCallbackToFinish.countDown();
-      callbackThread.join(TimeUnit.SECONDS.toMillis(5));
-      closeThread.join(TimeUnit.SECONDS.toMillis(5));
+      callbackThread.join(TimeUnit.SECONDS.toMillis(30));
+      closeThread.join(TimeUnit.SECONDS.toMillis(30));
       Assert.assertFalse(callbackThread.isAlive());
       Assert.assertFalse(closeThread.isAlive());
       Assert.assertTrue(sink.isClosed());
@@ -337,6 +337,8 @@ public void clearEventsReferenceCount() {
       allowCallbackToFinish.countDown();
       callbackThread.interrupt();
       closeThread.interrupt();
+      callbackThread.join(TimeUnit.SECONDS.toMillis(30));
+      closeThread.join(TimeUnit.SECONDS.toMillis(30));
     }
   }
 
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/DisruptorShutdownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/DisruptorShutdownTest.java
index ef57ea625aef6..7d18ddf7d3f7b 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/DisruptorShutdownTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/DisruptorShutdownTest.java
@@ -45,13 +45,15 @@ public void testBatchEventProcessorDrainsPublishedEventsOnShutdownInterrupt() th
             (event, sequence, endOfBatch) -> handledEventCount.incrementAndGet());
 
     final Thread processorThread = new Thread(processor, "pipe-batch-event-processor-test");
-    processorThread.start();
-
-    Assert.assertTrue(barrier.awaitWaitForCall());
-    processor.halt();
-    barrier.interruptWait();
-
-    processorThread.join(TimeUnit.SECONDS.toMillis(5));
+    processorThread.setDaemon(true);
+    try {
+      processorThread.start();
+      Assert.assertTrue(barrier.awaitWaitForCall());
+    } finally {
+      processor.halt();
+      barrier.interruptWait();
+      processorThread.join(TimeUnit.SECONDS.toMillis(30));
+    }
 
     Assert.assertFalse(processorThread.isAlive());
     Assert.assertEquals(1, handledEventCount.get());
@@ -83,12 +85,18 @@ public void testBatchEventProcessorDrainsEventsPublishedAfterCurrentBatchWhenHal
 
     final Thread processorThread =
         new Thread(processor, "pipe-batch-event-processor-snapshot-test");
-    processorThread.start();
-    processorThread.join(TimeUnit.SECONDS.toMillis(5));
-
-    Assert.assertFalse(processorThread.isAlive());
-    Assert.assertEquals(2, handledEventCount.get());
-    Assert.assertEquals(1L, processor.getSequence().get());
+    processorThread.setDaemon(true);
+    try {
+      processorThread.start();
+      processorThread.join(TimeUnit.SECONDS.toMillis(30));
+      Assert.assertFalse(processorThread.isAlive());
+      Assert.assertEquals(2, handledEventCount.get());
+      Assert.assertEquals(1L, processor.getSequence().get());
+    } finally {
+      processor.halt();
+      processorThread.interrupt();
+      processorThread.join(TimeUnit.SECONDS.toMillis(30));
+    }
   }
 
   @Test
@@ -97,6 +105,7 @@ public void testDisruptorShutdownInterruptsWaitingProcessor() throws Exception {
     final ThreadFactory threadFactory =
         runnable -> {
           final Thread thread = new Thread(runnable, "pipe-disruptor-shutdown-test");
+          thread.setDaemon(true);
           processorThreadReference.set(thread);
           return thread;
         };
@@ -106,10 +115,12 @@ public void testDisruptorShutdownInterruptsWaitingProcessor() throws Exception {
     disruptor.start();
 
     final Thread processorThread = processorThreadReference.get();
-    Assert.assertNotNull(processorThread);
-
-    TimeUnit.MILLISECONDS.sleep(50);
-    disruptor.shutdown();
+    try {
+      Assert.assertNotNull(processorThread);
+      TimeUnit.MILLISECONDS.sleep(50);
+    } finally {
+      disruptor.shutdown();
+    }
 
     Assert.assertFalse(processorThread.isAlive());
   }
@@ -120,6 +131,7 @@ public void testUnexpectedInterruptDoesNotStopProcessor() throws Exception {
     final ThreadFactory threadFactory =
         runnable -> {
           final Thread thread = new Thread(runnable, "pipe-disruptor-unexpected-interrupt-test");
+          thread.setDaemon(true);
           processorThreadReference.set(thread);
           return thread;
         };
@@ -130,16 +142,17 @@ public void testUnexpectedInterruptDoesNotStopProcessor() throws Exception {
         disruptor.handleEventsWith((event, sequence, endOfBatch) -> handled.countDown()).start();
 
     final Thread processorThread = processorThreadReference.get();
-    Assert.assertNotNull(processorThread);
-
-    TimeUnit.MILLISECONDS.sleep(50);
-    processorThread.interrupt();
-
-    ringBuffer.publishEvent((event, sequence, value) -> event.value = value, 1);
-    Assert.assertTrue(handled.await(5, TimeUnit.SECONDS));
-    Assert.assertTrue(processorThread.isAlive());
-
-    disruptor.shutdown();
+    try {
+      Assert.assertNotNull(processorThread);
+      TimeUnit.MILLISECONDS.sleep(50);
+      processorThread.interrupt();
+
+      ringBuffer.publishEvent((event, sequence, value) -> event.value = value, 1);
+      Assert.assertTrue(handled.await(30, TimeUnit.SECONDS));
+      Assert.assertTrue(processorThread.isAlive());
+    } finally {
+      disruptor.shutdown();
+    }
     Assert.assertFalse(processorThread.isAlive());
   }
 
@@ -160,10 +173,14 @@ public void testPublishEventCanAbortWhenClosingWhileBufferIsFull() throws Except
                         (event, sequence, value) -> event.value = value, 2, isClosed::get)),
             "pipe-disruptor-publish-abort-test");
 
-    publisherThread.start();
-    TimeUnit.MILLISECONDS.sleep(50);
-    isClosed.set(true);
-    publisherThread.join(TimeUnit.SECONDS.toMillis(5));
+    publisherThread.setDaemon(true);
+    try {
+      publisherThread.start();
+      TimeUnit.MILLISECONDS.sleep(50);
+    } finally {
+      isClosed.set(true);
+      publisherThread.join(TimeUnit.SECONDS.toMillis(30));
+    }
 
     Assert.assertFalse(publisherThread.isAlive());
     Assert.assertFalse(published.get());
@@ -202,7 +219,7 @@ public long getHighestPublishedSequence(final long lowerBound, final long availa
     }
 
     private boolean awaitWaitForCall() throws InterruptedException {
-      return waitForCalled.await(5, TimeUnit.SECONDS);
+      return waitForCalled.await(30, TimeUnit.SECONDS);
     }
 
     private void interruptWait() {