This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/fix-pipe-region-downgrading-retry in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9dd346bc6edcbbea250b8b11010487de316ab842 Author: Caideyipi <[email protected]> AuthorDate: Mon Nov 2 13:09:25 2026 +0800 Fix realtime TsFile source retry progress --- .../PipeRealtimeDataRegionTsFileSource.java | 31 ++++++++++--- .../db/pipe/source/PipeRealtimeExtractTest.java | 51 ++++++++++++++++++++++ 2 files changed, 75 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java index 14db05e678c..ec308b65d07 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java @@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory; import java.util.ArrayDeque; import java.util.Arrays; import java.util.Deque; +import java.util.Iterator; import java.util.PriorityQueue; import java.util.Queue; @@ -177,7 +178,7 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo public Event supply() { if (isRegionLevelDowngradingEnabled) { synchronized (tsFileOrderingLock) { - if (isClosed.get() || inFlightTsFileEpoch != null) { + if (isClosed.get()) { return null; } @@ -191,21 +192,37 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo bufferedEvents.offerLast(pendingEvent); } } - return supplyInternal(); + // A heartbeat may trigger retry of the in-flight TsFile at the sink. Keep files and other + // events ordered behind it, but do not block those retries while waiting for its commit. + return supplyInternal(inFlightTsFileEpoch != null); } } - return supplyInternal(); + return supplyInternal(false); } - private PipeRealtimeEvent pollNextEvent() { + private PipeRealtimeEvent pollNextEvent(final boolean heartbeatOnly) { if (!isRegionLevelDowngradingEnabled) { return (PipeRealtimeEvent) pendingQueue.directPoll(); } + if (heartbeatOnly) { + final Iterator<PipeRealtimeEvent> iterator = bufferedEvents.iterator(); + while (iterator.hasNext()) { + final PipeRealtimeEvent event = iterator.next(); + if (event.getEvent() instanceof PipeHeartbeatEvent) { + if (((PipeHeartbeatEvent) event.getEvent()).isCompletionBarrier()) { + return null; + } + iterator.remove(); + return event; + } + } + return null; + } return orderedTsFileEvents.isEmpty() ? bufferedEvents.pollFirst() : orderedTsFileEvents.poll(); } - private Event supplyInternal() { - PipeRealtimeEvent realtimeEvent = pollNextEvent(); + private Event supplyInternal(final boolean heartbeatOnly) { + PipeRealtimeEvent realtimeEvent = pollNextEvent(heartbeatOnly); while (realtimeEvent != null) { Event suppliedEvent = null; @@ -255,7 +272,7 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo return suppliedEvent; } - realtimeEvent = pollNextEvent(); + realtimeEvent = pollNextEvent(heartbeatOnly); } // means the pending queue is empty. diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java index c58cf516d82..6bf89d7a16a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java @@ -1333,6 +1333,57 @@ public class PipeRealtimeExtractTest { } } + @Test + public void testTsFileSourceRegionLevelDowngradingSuppliesHeartbeatWhileTsFileIsInFlight() + throws Exception { + final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1); + try (final PipeRealtimeDataRegionTsFileSource extractor = + createTsFileSourceForTest( + PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY, pipeTaskMeta)) { + final PipeRealtimeEvent firstEvent = + extractTsFileForTest( + extractor, + pipeTaskMeta, + createTsFileResource(dataRegion1, "123-1-0-0.tsfile"), + false); + final PipeRealtimeEvent secondEvent = + extractTsFileForTest( + extractor, + pipeTaskMeta, + createTsFileResource(dataRegion1, "123-2-0-0.tsfile"), + false); + final PipeRealtimeEvent heartbeat = + PipeRealtimeEventFactory.createRealtimeEvent(dataRegion1, false); + Assert.assertTrue(heartbeat.increaseReferenceCount(TEST_REFERENCE_HOLDER)); + extractor.extract(heartbeat); + final PipeRealtimeEvent completionBarrier = + PipeRealtimeEventFactory.createCompletionBarrierEvent(dataRegion1); + Assert.assertTrue(completionBarrier.increaseReferenceCount(TEST_REFERENCE_HOLDER)); + extractor.extract(completionBarrier); + final PipeRealtimeEvent heartbeatAfterCompletionBarrier = + PipeRealtimeEventFactory.createRealtimeEvent(dataRegion1, false); + Assert.assertTrue( + heartbeatAfterCompletionBarrier.increaseReferenceCount(TEST_REFERENCE_HOLDER)); + extractor.extract(heartbeatAfterCompletionBarrier); + + Assert.assertSame(firstEvent.getEvent(), extractor.supply()); + Assert.assertSame(heartbeat.getEvent(), extractor.supply()); + Assert.assertNull(extractor.supply()); + Assert.assertFalse(secondEvent.getEvent().isReleased()); + Assert.assertFalse(completionBarrier.getEvent().isReleased()); + Assert.assertFalse(heartbeatAfterCompletionBarrier.getEvent().isReleased()); + + releaseSuppliedEvent(heartbeat.getEvent()); + releaseSuppliedEvent(firstEvent.getEvent()); + Assert.assertSame(secondEvent.getEvent(), extractor.supply()); + releaseSuppliedEvent(secondEvent.getEvent()); + Assert.assertSame(completionBarrier.getEvent(), extractor.supply()); + releaseSuppliedEvent(completionBarrier.getEvent()); + Assert.assertSame(heartbeatAfterCompletionBarrier.getEvent(), extractor.supply()); + releaseSuppliedEvent(heartbeatAfterCompletionBarrier.getEvent()); + } + } + @Test public void testTsFileSourceRegionLevelDowngradingFollowsQueryPriority() throws Exception { final PipeEventCommitManager commitManager = PipeEventCommitManager.getInstance();
