From 6ff4d70c77bb39b2e54c4ededc3030d3493b1e14 Mon Sep 17 00:00:00 2001
From: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
Date: Fri, 18 Sep 2026 17:11:20 +0800
Subject: [PATCH] Fix SinkChannel failure propagation test race

---
 .../exchange/SinkChannelFailurePropagationTest.java  | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SinkChannelFailurePropagationTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SinkChannelFailurePropagationTest.java
index 9564b10476644..7030b5c210c2d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SinkChannelFailurePropagationTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SinkChannelFailurePropagationTest.java
@@ -308,20 +308,21 @@ public void testSingleChannelFailurePropagation() throws Exception {
       stateTracker.start();
 
       Assert.assertTrue(sinkHandle.isFull().isDone());
-      long failureStartNanos = System.nanoTime();
       sinkHandle.send(Utils.createMockTsBlocks(1, 1024).get(0));
 
       Mockito.verify(exchangeClient, Mockito.timeout(5_000).times(SinkChannel.MAX_ATTEMPT_TIMES))
           .onNewDataBlockEvent(Mockito.any(TNewDataBlockEvent.class));
+      long waitStartNanos = System.nanoTime();
       while (!fragmentStateMachine.getState().isDone()
-          && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - failureStartNanos) < 5) {
+          && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - waitStartNanos) < 5) {
         Thread.sleep(10);
       }
 
       Assert.assertEquals(FragmentInstanceState.FAILED, fragmentStateMachine.getState());
       Assert.assertEquals(expectedFailure, fragmentStateMachine.getFailureCauses().peek());
-      while (!sinkChannel.isAborted()
-          && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - failureStartNanos) < 5) {
+      waitStartNanos = System.nanoTime();
+      while (!sinkHandle.isAborted()
+          && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - waitStartNanos) < 5) {
         Thread.sleep(10);
       }
       Assert.assertTrue(sinkHandle.isAborted());
@@ -330,8 +331,9 @@ public void testSingleChannelFailurePropagation() throws Exception {
       Assert.assertFalse(sinkChannel.isClosed());
       Assert.assertEquals(0, exchangeManager.getShuffleSinkHandleSize());
 
+      waitStartNanos = System.nanoTime();
       while (!queryStateMachine.getState().isDone()
-          && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - failureStartNanos) < 5) {
+          && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - waitStartNanos) < 5) {
         Thread.sleep(10);
       }
       Mockito.verify(stateClient, Mockito.timeout(3_000).atLeastOnce())