From 3ceb360dde241adb2e94f9b13dfba9e6203d293c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:17:31 +0800 Subject: [PATCH 1/3] Fix snapshot pipe startup before IoTConsensus catch-up --- .../consensus/iot/IoTConsensusServerImpl.java | 9 ++++++ .../dataregion/IoTDBDataRegionSource.java | 30 ++++++++++++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index e074e7204ee51..277859706750a 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -812,6 +812,15 @@ public void waitTargetPeerUntilSyncLogCompleted(Peer targetPeer) } } + /** Wait until every remote peer has synchronized its writes to all peers. */ + public void waitUntilAllRemotePeersSyncLogCompleted() throws ConsensusGroupModifyPeerException { + for (Peer peer : getConfiguration()) { + if (!peer.equals(thisNode)) { + waitTargetPeerUntilSyncLogCompleted(peer); + } + } + } + public boolean hasReleaseAllRegionRelatedResource(ConsensusGroupId groupId) { return stateMachine.hasReleaseAllRegionRelatedResource(groupId); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java index 5d677e100472f..132fe92f306ae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java @@ -27,6 +27,10 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; import org.apache.iotdb.commons.pipe.source.IoTDBSource; import org.apache.iotdb.commons.queryengine.common.SqlDialect; +import org.apache.iotdb.consensus.IConsensus; +import org.apache.iotdb.consensus.iot.IoTConsensus; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; @@ -536,10 +540,12 @@ public void start() throws Exception { historicalSource.getClass().getSimpleName(), realtimeSource.getClass().getSimpleName()); + final DataRegionId dataRegionIdObject = new DataRegionId(this.regionId); + waitUntilAllRemotePeersSyncLogCompletedIfNecessary(dataRegionIdObject); + super.start(); final AtomicReference exceptionHolder = new AtomicReference<>(null); - final DataRegionId dataRegionIdObject = new DataRegionId(this.regionId); while (true) { // try to start sources in the data region ... // first try to run if data region exists, then try to run if data region does not exist. @@ -598,6 +604,28 @@ private void startHistoricalExtractorAndRealtimeExtractor( } } + private void waitUntilAllRemotePeersSyncLogCompletedIfNecessary(final DataRegionId dataRegionId) + throws Exception { + // A heartbeat-only source cannot capture writes that arrive after historical scanning. Before + // starting that scan, wait for every remote IoTConsensus writer to finish applying its pending + // writes to this local replica. This method must be called without holding the DataRegion write + // lock because applying a remote write may need the same lock. + if (!(realtimeSource instanceof PipeRealtimeDataRegionHeartbeatSource)) { + return; + } + + final IConsensus dataRegionConsensus = DataRegionConsensusImpl.getInstance(); + if (!(dataRegionConsensus instanceof IoTConsensus)) { + return; + } + + final IoTConsensusServerImpl consensusServer = + ((IoTConsensus) dataRegionConsensus).getImpl(dataRegionId); + if (consensusServer != null) { + consensusServer.waitUntilAllRemotePeersSyncLogCompleted(); + } + } + private void rethrowExceptionIfAny(final AtomicReference exceptionHolder) { if (exceptionHolder.get() != null) { throw new PipeException(DataNodePipeMessages.FAILED_TO_START_SOURCES, exceptionHolder.get()); From 6496ceeaee557e901489030f4a39caecfb6e1ad1 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 31 Aug 2026 18:28:36 +0800 Subject: [PATCH 2/3] Stabilize pipe subtask executor test --- .../db/pipe/agent/task/PipeSubtaskExecutorTest.java | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java index 892c61536e17d..da3a8349c439b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java @@ -26,8 +26,8 @@ import org.junit.Assert; import org.junit.Test; -import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.times; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; public abstract class PipeSubtaskExecutorTest { @@ -71,12 +71,7 @@ public void testStart() throws Exception { // test start a subtask which is in the map executor.register(subtask); executor.start(subtask.getTaskID()); - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - verify(subtask, atLeast(2)).call(); + verify(subtask, timeout(10_000).atLeast(2)).call(); Assert.assertTrue(subtask.isSubmittingSelf()); // test start a subtask which is in the map and is already running From 90b73c2189b1369bad89bcf0cd0367077a854fc8 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:14:47 +0800 Subject: [PATCH 3/3] Update PipeSubtaskExecutorTest.java --- .../iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java index da3a8349c439b..ae0450a034414 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeSubtaskExecutorTest.java @@ -26,8 +26,8 @@ import org.junit.Assert; import org.junit.Test; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; public abstract class PipeSubtaskExecutorTest {