Fix snapshot pipe startup before IoTConsensus catch-up

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);
   }

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<Exception> 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<Exception> exceptionHolder) {
     if (exceptionHolder.get() != null) {
       throw new PipeException(DataNodePipeMessages.FAILED_TO_START_SOURCES, exceptionHolder.get());