IoTV2: Refactor replicate index so that it is shared at the pipe task level & Add some log for delete local peer (apache#15815)

Pengzna · web-flow · commit b8fd08984cb2 · 2025-06-26T09:33:48.000+08:00
* use custom replicate index for each consensus pipe * test conf * fix * Revert "test conf" This reverts commit f0f13af. * fix review
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
@@ -375,13 +375,14 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException
         if (!stateMachineMap.containsKey(groupId)) {
           throw new ConsensusGroupNotExistException(groupId);
         }
-
+        LOGGER.info("[{}] start to delete local peer for group {}", CLASS_NAME, groupId);
         final PipeConsensusServerImpl consensus = stateMachineMap.get(groupId);
         consensus.clear();
         stateMachineMap.remove(groupId);
 
         FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId)));
         KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
+        LOGGER.info("[{}] finish deleting local peer for group {}", CLASS_NAME, groupId);
       } finally {
         stateMachineMapLock.readLock().unlock();
       }
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ConsensusPipeConnector.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ConsensusPipeConnector.java
@@ -19,7 +19,7 @@
 package org.apache.iotdb.consensus.pipe.consensuspipe;
 
 public interface ConsensusPipeConnector {
-  long getConsensusPipeCommitProgress();
+  long getLeaderReplicateProgress();
 
-  long getConsensusPipeReplicateProgress();
+  long getFollowerApplyProgress();
 }
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ReplicateProgressManager.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ReplicateProgressManager.java
@@ -32,6 +32,6 @@ public interface ReplicateProgressManager {
   long getSyncLagForSpecificConsensusPipe(
       ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName);
 
-  void pinCommitIndexForMigration(
+  void pinReplicateIndexForRegionMigration(
       ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName);
 }
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/PipeConsensusSyncLagManager.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/PipeConsensusSyncLagManager.java
@@ -48,9 +48,7 @@ public long getSyncLagForRegionMigration(
     return Optional.ofNullable(consensusPipe2ConnectorMap.get(consensusPipeName))
         .map(
             consensusPipeConnector ->
-                Math.max(
-                    pinnedCommitIndex - consensusPipeConnector.getConsensusPipeReplicateProgress(),
-                    0L))
+                Math.max(pinnedCommitIndex - consensusPipeConnector.getFollowerApplyProgress(), 0L))
         .orElse(0L);
   }
 
@@ -62,16 +60,16 @@ public long getSyncLagForSpecificConsensusPipe(ConsensusPipeName consensusPipeNa
     return Optional.ofNullable(consensusPipe2ConnectorMap.get(consensusPipeName))
         .map(
             consensusPipeConnector -> {
-              long userWriteProgress = consensusPipeConnector.getConsensusPipeCommitProgress();
-              long replicateProgress = consensusPipeConnector.getConsensusPipeReplicateProgress();
+              long userWriteProgress = consensusPipeConnector.getLeaderReplicateProgress();
+              long replicateProgress = consensusPipeConnector.getFollowerApplyProgress();
               return Math.max(userWriteProgress - replicateProgress, 0L);
             })
         .orElse(0L);
   }
 
-  public long getCurrentCommitIndex(ConsensusPipeName consensusPipeName) {
+  public long getCurrentLeaderReplicateIndex(ConsensusPipeName consensusPipeName) {
     return Optional.ofNullable(consensusPipe2ConnectorMap.get(consensusPipeName))
-        .map(ConsensusPipeConnector::getConsensusPipeCommitProgress)
+        .map(ConsensusPipeConnector::getLeaderReplicateProgress)
         .orElse(0L);
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtaskManager.java
@@ -31,6 +31,7 @@
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.pipe.agent.task.execution.PipeConnectorSubtaskExecutor;
 import org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBDataRegionAsyncConnector;
+import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
 import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
 import org.apache.iotdb.db.storageengine.StorageEngine;
 import org.apache.iotdb.pipe.api.PipeConnector;
@@ -201,6 +202,9 @@ public synchronized void deregister(
     }
 
     PipeEventCommitManager.getInstance().deregister(pipeName, creationTime, regionId);
+    // Reset IoTV2 replicate index to prevent index jumps. Do this when a consensus pipe no longer
+    // replicates data, since extractor and processor are already dropped now.
+    ReplicateProgressDataNodeManager.resetReplicateIndexForIoTV2(pipeName);
   }
 
   public synchronized void start(final String attributeSortedString) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java
@@ -28,7 +28,6 @@
 import org.apache.iotdb.commons.consensus.ConsensusGroupId;
 import org.apache.iotdb.commons.consensus.index.ProgressIndex;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorRetryTimesConfigurableException;
-import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
 import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.commons.service.metric.MetricService;
@@ -39,7 +38,6 @@
 import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusDeleteEventHandler;
 import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletBatchEventHandler;
 import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletInsertNodeEventHandler;
@@ -48,6 +46,7 @@
 import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusDeleteNodeReq;
 import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletBinaryReq;
 import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletInsertNodeReq;
+import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
 import org.apache.iotdb.db.pipe.consensus.metric.PipeConsensusConnectorMetrics;
 import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
@@ -726,16 +725,12 @@ public int getRetryBufferSize() {
   }
 
   @Override
-  public long getConsensusPipeCommitProgress() {
-    return PipeEventCommitManager.getInstance()
-        .getGivenConsensusPipeCommitId(
-            consensusPipeName,
-            PipeDataNodeAgent.task().getPipeCreationTime(consensusPipeName),
-            consensusGroupId);
+  public long getLeaderReplicateProgress() {
+    return ReplicateProgressDataNodeManager.getReplicateIndexForIoTV2(consensusPipeName);
   }
 
   @Override
-  public long getConsensusPipeReplicateProgress() {
+  public long getFollowerApplyProgress() {
     return currentReplicateProgress;
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ReplicateProgressDataNodeManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ReplicateProgressDataNodeManager.java
@@ -44,7 +44,8 @@
 
 public class ReplicateProgressDataNodeManager implements ReplicateProgressManager {
   private static final int DATA_NODE_ID = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
-  private static final Map<String, AtomicLong> groupId2ReplicateIndex = new ConcurrentHashMap<>();
+  private static final Map<String, AtomicLong> consensusPipe2ReplicateIndex =
+      new ConcurrentHashMap<>();
   private final Map<ConsensusGroupId, ProgressIndex> groupId2MaxProgressIndex;
   private final Map<ConsensusPipeName, Long> consensusPipe2pinnedCommitIndexForMigration;
 
@@ -55,12 +56,20 @@ public ReplicateProgressDataNodeManager() {
     recoverMaxProgressIndexFromDataRegion();
   }
 
-  public static long assignReplicateIndexForIoTV2(String groupId) {
-    return groupId2ReplicateIndex
-        .compute(groupId, (k, v) -> v == null ? new AtomicLong(0) : v)
+  public static long assignReplicateIndexForIoTV2(String consensusPipeName) {
+    return consensusPipe2ReplicateIndex
+        .compute(consensusPipeName, (k, v) -> v == null ? new AtomicLong(0) : v)
         .incrementAndGet();
   }
 
+  public static void resetReplicateIndexForIoTV2(String consensusPipeName) {
+    consensusPipe2ReplicateIndex.put(consensusPipeName, new AtomicLong(0));
+  }
+
+  public static long getReplicateIndexForIoTV2(String consensusPipeName) {
+    return consensusPipe2ReplicateIndex.getOrDefault(consensusPipeName, new AtomicLong(0)).get();
+  }
+
   public static ProgressIndex extractLocalSimpleProgressIndex(ProgressIndex progressIndex) {
     if (progressIndex instanceof RecoverProgressIndex) {
       final Map<Integer, SimpleProgressIndex> dataNodeId2LocalIndex =
@@ -151,11 +160,11 @@ public long getSyncLagForSpecificConsensusPipe(
   }
 
   @Override
-  public void pinCommitIndexForMigration(
+  public void pinReplicateIndexForRegionMigration(
       ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName) {
     this.consensusPipe2pinnedCommitIndexForMigration.put(
         consensusPipeName,
         PipeConsensusSyncLagManager.getInstance(consensusGroupId.toString())
-            .getCurrentCommitIndex(consensusPipeName));
+            .getCurrentLeaderReplicateIndex(consensusPipeName));
   }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
@@ -20,34 +20,24 @@
 package org.apache.iotdb.db.pipe.event.realtime;
 
 import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
-import org.apache.iotdb.consensus.pipe.PipeConsensus;
-import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
-import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
 import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
 import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.epoch.TsFileEpochManager;
-import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.DeviceIDFactory;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryHandler;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.stream.Collectors;
 
 public class PipeRealtimeEventFactory {
-
-  private static final Logger LOGGER = LoggerFactory.getLogger(PipeRealtimeEventFactory.class);
   private static final TsFileEpochManager TS_FILE_EPOCH_MANAGER = new TsFileEpochManager();
 
   public static PipeRealtimeEvent createRealtimeEvent(
-      final String dataRegionId,
       final Boolean isTableModel,
       final String databaseNameFromDataRegion,
       final TsFileResource resource,
@@ -56,23 +46,10 @@ public static PipeRealtimeEvent createRealtimeEvent(
         new PipeTsFileInsertionEvent(
             isTableModel, databaseNameFromDataRegion, resource, isLoaded, false);
 
-    // if using IoTV2, assign a replicateIndex for this event
-    if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
-        && PipeConsensusProcessor.isShouldReplicate(tsFileInsertionEvent)) {
-      tsFileInsertionEvent.setReplicateIndexForIoTV2(
-          ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
-      LOGGER.info(
-          "[Region{}]Set {} for event {}",
-          dataRegionId,
-          tsFileInsertionEvent.getReplicateIndexForIoTV2(),
-          tsFileInsertionEvent);
-    }
-
     return TS_FILE_EPOCH_MANAGER.bindPipeTsFileInsertionEvent(tsFileInsertionEvent, resource);
   }
 
   public static PipeRealtimeEvent createRealtimeEvent(
-      final String dataRegionId,
       final Boolean isTableModel,
       final String databaseNameFromDataRegion,
       final WALEntryHandler walEntryHandler,
@@ -98,18 +75,6 @@ public static PipeRealtimeEvent createRealtimeEvent(
             insertNode.isAligned(),
             insertNode.isGeneratedByPipe());
 
-    // if using IoTV2, assign a replicateIndex for this event
-    if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
-        && PipeConsensusProcessor.isShouldReplicate(insertionEvent)) {
-      insertionEvent.setReplicateIndexForIoTV2(
-          ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
-      LOGGER.info(
-          "[Region{}]Set {} for event {}",
-          dataRegionId,
-          insertionEvent.getReplicateIndexForIoTV2(),
-          insertionEvent);
-    }
-
     return TS_FILE_EPOCH_MANAGER.bindPipeInsertNodeTabletInsertionEvent(
         insertionEvent, insertNode, resource);
   }
@@ -120,23 +85,10 @@ public static PipeRealtimeEvent createRealtimeEvent(
         new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null);
   }
 
-  public static PipeRealtimeEvent createRealtimeEvent(
-      final String dataRegionId, final AbstractDeleteDataNode node) {
+  public static PipeRealtimeEvent createRealtimeEvent(final AbstractDeleteDataNode node) {
     PipeDeleteDataNodeEvent deleteDataNodeEvent =
         new PipeDeleteDataNodeEvent(node, node.isGeneratedByPipe());
 
-    // if using IoTV2, assign a replicateIndex for this event
-    if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
-        && PipeConsensusProcessor.isShouldReplicate(deleteDataNodeEvent)) {
-      deleteDataNodeEvent.setReplicateIndexForIoTV2(
-          ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
-      LOGGER.info(
-          "[Region{}]Set {} for event {}",
-          dataRegionId,
-          deleteDataNodeEvent.getReplicateIndexForIoTV2(),
-          deleteDataNodeEvent);
-    }
-
     return new PipeRealtimeEvent(deleteDataNodeEvent, null, null);
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
@@ -877,13 +877,9 @@ private Event supplyTsFileEvent(final TsFileResource resource) {
     if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
         && PipeConsensusProcessor.isShouldReplicate(event)) {
       event.setReplicateIndexForIoTV2(
-          ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(
-              resource.getDataRegionId()));
+          ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(pipeName));
       LOGGER.info(
-          "[Region{}]Set {} for event {}",
-          resource.getDataRegionId(),
-          event.getReplicateIndexForIoTV2(),
-          event);
+          "[{}]Set {} for historical event {}", pipeName, event.getReplicateIndexForIoTV2(), event);
     }
 
     if (sloppyPattern || isDbNameCoveredByPattern) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -27,6 +27,9 @@
 import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
 import org.apache.iotdb.commons.pipe.metric.PipeEventCounter;
 import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.consensus.pipe.PipeConsensus;
+import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
+import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
 import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
@@ -40,6 +43,7 @@
 import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.matcher.PipeDataRegionMatcher;
 import org.apache.iotdb.db.pipe.metric.source.PipeAssignerMetrics;
 import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
+import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor;
 import org.apache.iotdb.db.storageengine.StorageEngine;
 import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -191,8 +195,18 @@ private void assignToExtractor(
                       extractor.getRealtimeDataExtractionStartTime(),
                       extractor.getRealtimeDataExtractionEndTime());
               final EnrichedEvent innerEvent = copiedEvent.getEvent();
-              // Bind replicateIndex for IoTV2
-              innerEvent.setReplicateIndexForIoTV2(event.getEvent().getReplicateIndexForIoTV2());
+              // if using IoTV2, assign a replicateIndex for this realtime event
+              if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
+                  && PipeConsensusProcessor.isShouldReplicate(innerEvent)) {
+                innerEvent.setReplicateIndexForIoTV2(
+                    ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(
+                        extractor.getPipeName()));
+                LOGGER.info(
+                    "[{}]Set {} for realtime event {}",
+                    extractor.getPipeName(),
+                    innerEvent.getReplicateIndexForIoTV2(),
+                    innerEvent);
+              }
 
               if (innerEvent instanceof PipeTsFileInsertionEvent) {
                 final PipeTsFileInsertionEvent tsFileInsertionEvent =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
@@ -116,7 +116,7 @@ public void listenToTsFile(
 
     assigner.publishToAssign(
         PipeRealtimeEventFactory.createRealtimeEvent(
-            dataRegionId, assigner.isTableModel(), databaseName, tsFileResource, isLoaded));
+            assigner.isTableModel(), databaseName, tsFileResource, isLoaded));
   }
 
   public void listenToInsertNode(
@@ -138,12 +138,7 @@ public void listenToInsertNode(
 
     assigner.publishToAssign(
         PipeRealtimeEventFactory.createRealtimeEvent(
-            dataRegionId,
-            assigner.isTableModel(),
-            databaseName,
-            walEntryHandler,
-            insertNode,
-            tsFileResource));
+            assigner.isTableModel(), databaseName, walEntryHandler, insertNode, tsFileResource));
   }
 
   public DeletionResource listenToDeleteData(
@@ -169,7 +164,7 @@ public DeletionResource listenToDeleteData(
       deletionResource = null;
     }
 
-    assigner.publishToAssign(PipeRealtimeEventFactory.createRealtimeEvent(regionId, node));
+    assigner.publishToAssign(PipeRealtimeEventFactory.createRealtimeEvent(node));
 
     return deletionResource;
   }
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java
@@ -176,17 +176,6 @@ public void setCommitRateMarker(final BiConsumer<String, Boolean> commitRateMark
     this.commitRateMarker = commitRateMarker;
   }
 
-  public long getGivenConsensusPipeCommitId(
-      final String consensusPipeName, final long creationTime, final int consensusGroupId) {
-    final CommitterKey committerKey =
-        generateCommitterKey(consensusPipeName, creationTime, consensusGroupId);
-    final PipeEventCommitter committer = eventCommitterMap.get(committerKey);
-    if (committer == null) {
-      return 0;
-    }
-    return committer.getCurrentCommitId();
-  }
-
   //////////////////////////// singleton ////////////////////////////
 
   private PipeEventCommitManager() {

Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,7 @@`
`19`	`19`	`package org.apache.iotdb.consensus.pipe.consensuspipe;`
`20`	`20`
`21`	`21`	`public interface ConsensusPipeConnector {`
`22`		`- long getConsensusPipeCommitProgress();`
	`22`	`+ long getLeaderReplicateProgress();`
`23`	`23`
`24`		`- long getConsensusPipeReplicateProgress();`
	`24`	`+ long getFollowerApplyProgress();`
`25`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,6 @@ public interface ReplicateProgressManager {`
`32`	`32`	`long getSyncLagForSpecificConsensusPipe(`
`33`	`33`	`ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName);`
`34`	`34`
`35`		`- void pinCommitIndexForMigration(`
	`35`	`+ void pinReplicateIndexForRegionMigration(`
`36`	`36`	`ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName);`
`37`	`37`	`}`