This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new fa9711e1146 IoTConsensusV2: Use custom commitIndex to indicate
synchronization progress instead of pipe's commitIndex (#14624)
fa9711e1146 is described below
commit fa9711e1146ad53a435fa6cf89bdf7aab2648ba7
Author: Peng Junzhi <[email protected]>
AuthorDate: Wed Feb 26 17:06:30 2025 +0800
IoTConsensusV2: Use custom commitIndex to indicate synchronization progress
instead of pipe's commitIndex (#14624)
* wip: use syncLag for region migration progress management
* wip: complete syncLag framwork. TODO: maintain progressIndex locally as
syncLag calculator
* spotless
* complete replicateIndex
* fix
* fix for region migration
* spotless
* test conf
* fix replicate failed
* Revert "test conf"
This reverts commit a7858f92707abe64d7c66a1220b966fe0c98c10b.
* remove duplicate & run ci
* test conf
* fix the order of tsfile
* Revert "test conf"
This reverts commit d128b84ed786385688f71867e2e369c46b24dd21.
* retrigger ci
* fix ci
* assign replicate id when flush
* fix
* test conf
* fix historical
* close log
* Revert "test conf"
This reverts commit 34158b0b9f92e15b2bdc42a3e3c565ac1dbf4e0e.
---
.../consensus/config/PipeConsensusConfig.java | 21 ++++---
.../consensus/pipe/PipeConsensusServerImpl.java | 21 +++++--
...xManager.java => ReplicateProgressManager.java} | 8 ++-
.../pipe/metric/PipeConsensusSyncLagManager.java | 71 +++++++++++++++++-----
.../db/consensus/DataRegionConsensusImpl.java | 4 +-
.../pipeconsensus/PipeConsensusAsyncConnector.java | 16 ++---
.../pipeconsensus/PipeConsensusSyncConnector.java | 9 +--
.../PipeConsensusTabletInsertionEventHandler.java | 8 +--
.../PipeConsensusTsFileInsertionEventHandler.java | 8 +--
.../PipeConsensusTransferBatchReqBuilder.java | 5 +-
....java => ReplicateProgressDataNodeManager.java} | 36 ++++++++++-
.../deletion/DeletionResourceManager.java | 4 +-
.../deletion/persist/PageCacheDeletionBuffer.java | 5 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 17 +++++-
.../event/realtime/PipeRealtimeEventFactory.java | 71 +++++++++++++++++++---
...oricalDataRegionTsFileAndDeletionExtractor.java | 17 ++++++
.../realtime/assigner/PipeDataRegionAssigner.java | 2 +
.../listener/PipeInsertionDataNodeListener.java | 16 ++++-
.../pipeconsensus/PipeConsensusProcessor.java | 43 +++++++++----
.../pipeconsensus/PipeConsensusReceiver.java | 55 +++--------------
.../iotdb/commons/pipe/event/EnrichedEvent.java | 17 +++++-
.../src/main/thrift/pipeconsensus.thrift | 5 +-
22 files changed, 316 insertions(+), 143 deletions(-)
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/PipeConsensusConfig.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/PipeConsensusConfig.java
index 349a5cd7e1d..eb6ea361d24 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/PipeConsensusConfig.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/PipeConsensusConfig.java
@@ -24,7 +24,7 @@ import
org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeDispatcher;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeGuardian;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeReceiver;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeSelector;
-import org.apache.iotdb.consensus.pipe.consensuspipe.ProgressIndexManager;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ReplicateProgressManager;
import java.util.concurrent.TimeUnit;
@@ -203,7 +203,7 @@ public class PipeConsensusConfig {
private final ConsensusPipeDispatcher consensusPipeDispatcher;
private final ConsensusPipeGuardian consensusPipeGuardian;
private final ConsensusPipeSelector consensusPipeSelector;
- private final ProgressIndexManager progressIndexManager;
+ private final ReplicateProgressManager replicateProgressManager;
private final ConsensusPipeReceiver consensusPipeReceiver;
private final long consensusPipeGuardJobIntervalInSeconds;
@@ -214,7 +214,7 @@ public class PipeConsensusConfig {
ConsensusPipeDispatcher consensusPipeDispatcher,
ConsensusPipeGuardian consensusPipeGuardian,
ConsensusPipeSelector consensusPipeSelector,
- ProgressIndexManager progressIndexManager,
+ ReplicateProgressManager replicateProgressManager,
ConsensusPipeReceiver consensusPipeReceiver,
long consensusPipeGuardJobIntervalInSeconds) {
this.extractorPluginName = extractorPluginName;
@@ -223,7 +223,7 @@ public class PipeConsensusConfig {
this.consensusPipeDispatcher = consensusPipeDispatcher;
this.consensusPipeGuardian = consensusPipeGuardian;
this.consensusPipeSelector = consensusPipeSelector;
- this.progressIndexManager = progressIndexManager;
+ this.replicateProgressManager = replicateProgressManager;
this.consensusPipeReceiver = consensusPipeReceiver;
this.consensusPipeGuardJobIntervalInSeconds =
consensusPipeGuardJobIntervalInSeconds;
}
@@ -256,8 +256,8 @@ public class PipeConsensusConfig {
return consensusPipeReceiver;
}
- public ProgressIndexManager getProgressIndexManager() {
- return progressIndexManager;
+ public ReplicateProgressManager getProgressIndexManager() {
+ return replicateProgressManager;
}
public long getConsensusPipeGuardJobIntervalInSeconds() {
@@ -277,7 +277,7 @@ public class PipeConsensusConfig {
private ConsensusPipeDispatcher consensusPipeDispatcher = null;
private ConsensusPipeGuardian consensusPipeGuardian = null;
private ConsensusPipeSelector consensusPipeSelector = null;
- private ProgressIndexManager progressIndexManager = null;
+ private ReplicateProgressManager replicateProgressManager = null;
private ConsensusPipeReceiver consensusPipeReceiver = null;
private long consensusPipeGuardJobIntervalInSeconds = 180L;
@@ -317,8 +317,9 @@ public class PipeConsensusConfig {
return this;
}
- public Pipe.Builder setProgressIndexManager(ProgressIndexManager
progressIndexManager) {
- this.progressIndexManager = progressIndexManager;
+ public Pipe.Builder setProgressIndexManager(
+ ReplicateProgressManager replicateProgressManager) {
+ this.replicateProgressManager = replicateProgressManager;
return this;
}
@@ -336,7 +337,7 @@ public class PipeConsensusConfig {
consensusPipeDispatcher,
consensusPipeGuardian,
consensusPipeSelector,
- progressIndexManager,
+ replicateProgressManager,
consensusPipeReceiver,
consensusPipeGuardJobIntervalInSeconds);
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
index 25a86ca5178..df1d6ff1f51 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
@@ -42,7 +42,7 @@ import
org.apache.iotdb.consensus.config.PipeConsensusConfig.ReplicateMode;
import org.apache.iotdb.consensus.exception.ConsensusGroupModifyPeerException;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeManager;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName;
-import org.apache.iotdb.consensus.pipe.consensuspipe.ProgressIndexManager;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ReplicateProgressManager;
import org.apache.iotdb.consensus.pipe.metric.PipeConsensusServerMetrics;
import org.apache.iotdb.consensus.pipe.thrift.TCheckConsensusPipeCompletedReq;
import org.apache.iotdb.consensus.pipe.thrift.TCheckConsensusPipeCompletedResp;
@@ -89,7 +89,7 @@ public class PipeConsensusServerImpl {
private final AtomicBoolean isStarted;
private final String consensusGroupId;
private final ConsensusPipeManager consensusPipeManager;
- private final ProgressIndexManager progressIndexManager;
+ private final ReplicateProgressManager replicateProgressManager;
private final IClientManager<TEndPoint, SyncPipeConsensusServiceClient>
syncClientManager;
private final PipeConsensusServerMetrics pipeConsensusServerMetrics;
private final ReplicateMode replicateMode;
@@ -111,7 +111,7 @@ public class PipeConsensusServerImpl {
this.isStarted = new AtomicBoolean(false);
this.consensusGroupId = thisNode.getGroupId().toString();
this.consensusPipeManager = consensusPipeManager;
- this.progressIndexManager = config.getPipe().getProgressIndexManager();
+ this.replicateProgressManager = config.getPipe().getProgressIndexManager();
this.syncClientManager = syncClientManager;
this.pipeConsensusServerMetrics = new PipeConsensusServerMetrics(this);
this.replicateMode = config.getReplicateMode();
@@ -319,7 +319,7 @@ public class PipeConsensusServerImpl {
long writeToStateMachineStartTime = System.nanoTime();
if (request instanceof ComparableConsensusRequest) {
((ComparableConsensusRequest) request)
-
.setProgressIndex(progressIndexManager.assignProgressIndex(thisNode.getGroupId()));
+
.setProgressIndex(replicateProgressManager.assignProgressIndex(thisNode.getGroupId()));
}
TSStatus result = stateMachine.write(request);
long writeToStateMachineEndTime = System.nanoTime();
@@ -608,12 +608,21 @@ public class PipeConsensusServerImpl {
}
}
+ public boolean isConsensusPipesTransmissionCompleted(List<String>
consensusPipeNames) {
+ return consensusPipeNames.stream()
+ .noneMatch(
+ pipeName ->
+ replicateProgressManager.getSyncLagForSpecificConsensusPipe(
+ thisNode.getGroupId(), new ConsensusPipeName(pipeName))
+ > 0);
+ }
+
public synchronized boolean isConsensusPipesTransmissionCompleted(
List<String> consensusPipeNames, boolean refreshCachedProgressIndex) {
if (refreshCachedProgressIndex) {
cachedProgressIndex =
cachedProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
-
progressIndexManager.getMaxAssignedProgressIndex(thisNode.getGroupId()));
+
replicateProgressManager.getMaxAssignedProgressIndex(thisNode.getGroupId()));
}
try {
@@ -621,7 +630,7 @@ public class PipeConsensusServerImpl {
.noneMatch(
name ->
cachedProgressIndex.isAfter(
- progressIndexManager.getProgressIndex(new
ConsensusPipeName(name))));
+ replicateProgressManager.getProgressIndex(new
ConsensusPipeName(name))));
} catch (PipeException e) {
LOGGER.info(e.getMessage());
return false;
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ProgressIndexManager.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ReplicateProgressManager.java
similarity index 81%
rename from
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ProgressIndexManager.java
rename to
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ReplicateProgressManager.java
index e5dc2e922ab..13a18132272 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ProgressIndexManager.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ReplicateProgressManager.java
@@ -22,10 +22,16 @@ package org.apache.iotdb.consensus.pipe.consensuspipe;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
-public interface ProgressIndexManager {
+public interface ReplicateProgressManager {
ProgressIndex getProgressIndex(ConsensusPipeName consensusPipeName);
ProgressIndex assignProgressIndex(ConsensusGroupId consensusGroupId);
ProgressIndex getMaxAssignedProgressIndex(ConsensusGroupId consensusGroupId);
+
+ long getSyncLagForSpecificConsensusPipe(
+ ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName);
+
+ void pinCommitIndexForMigration(
+ 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
index a65662d5f5e..7b047726a58 100644
---
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
@@ -20,10 +20,10 @@
package org.apache.iotdb.consensus.pipe.metric;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeConnector;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
@@ -36,27 +36,59 @@ import java.util.concurrent.locks.ReentrantLock;
public class PipeConsensusSyncLagManager {
long syncLag = Long.MIN_VALUE;
ReentrantLock lock = new ReentrantLock();
- List<ConsensusPipeConnector> consensusPipeConnectorList = new ArrayList<>();
+ Map<ConsensusPipeName, ConsensusPipeConnector> consensusPipe2ConnectorMap =
+ new ConcurrentHashMap<>();
- private long getSyncLagForSpecificConsensusPipe(ConsensusPipeConnector
consensusPipeConnector) {
- long userWriteProgress =
consensusPipeConnector.getConsensusPipeCommitProgress();
- long replicateProgress =
consensusPipeConnector.getConsensusPipeReplicateProgress();
- return Math.max(userWriteProgress - replicateProgress, 0);
+ /**
+ * pinnedCommitIndex - currentReplicateProgress. If res <= 0, indicating
that replication is
+ * finished.
+ */
+ public long getSyncLagForRegionMigration(
+ ConsensusPipeName consensusPipeName, long pinnedCommitIndex) {
+ return
Optional.ofNullable(consensusPipe2ConnectorMap.get(consensusPipeName))
+ .map(
+ consensusPipeConnector ->
+ Math.max(
+ pinnedCommitIndex -
consensusPipeConnector.getConsensusPipeReplicateProgress(),
+ 0L))
+ .orElse(0L);
+ }
+
+ /**
+ * userWriteProgress - currentReplicateProgress. If res <= 0, indicating
that replication is
+ * finished.
+ */
+ public long getSyncLagForSpecificConsensusPipe(ConsensusPipeName
consensusPipeName) {
+ return
Optional.ofNullable(consensusPipe2ConnectorMap.get(consensusPipeName))
+ .map(
+ consensusPipeConnector -> {
+ long userWriteProgress =
consensusPipeConnector.getConsensusPipeCommitProgress();
+ long replicateProgress =
consensusPipeConnector.getConsensusPipeReplicateProgress();
+ return Math.max(userWriteProgress - replicateProgress, 0L);
+ })
+ .orElse(0L);
+ }
+
+ public long getCurrentCommitIndex(ConsensusPipeName consensusPipeName) {
+ return
Optional.ofNullable(consensusPipe2ConnectorMap.get(consensusPipeName))
+ .map(ConsensusPipeConnector::getConsensusPipeCommitProgress)
+ .orElse(0L);
}
- public void addConsensusPipeConnector(ConsensusPipeConnector
consensusPipeConnector) {
+ public void addConsensusPipeConnector(
+ ConsensusPipeName consensusPipeName, ConsensusPipeConnector
consensusPipeConnector) {
try {
lock.lock();
- consensusPipeConnectorList.add(consensusPipeConnector);
+ consensusPipe2ConnectorMap.put(consensusPipeName,
consensusPipeConnector);
} finally {
lock.unlock();
}
}
- public void removeConsensusPipeConnector(ConsensusPipeConnector connector) {
+ public void removeConsensusPipeConnector(ConsensusPipeName
consensusPipeName) {
try {
lock.lock();
- consensusPipeConnectorList.remove(connector);
+ consensusPipe2ConnectorMap.remove(consensusPipeName);
} finally {
lock.unlock();
}
@@ -71,21 +103,27 @@ public class PipeConsensusSyncLagManager {
try {
lock.lock();
// if there isn't a consensus pipe task, the syncLag is 0
- if (consensusPipeConnectorList.isEmpty()) {
+ if (consensusPipe2ConnectorMap.isEmpty()) {
return 0;
}
// else we find the biggest gap between leader and replicas in all
consensus pipe task.
syncLag = Long.MIN_VALUE;
- consensusPipeConnectorList.forEach(
- consensusPipeConnector ->
- syncLag =
- Math.max(syncLag,
getSyncLagForSpecificConsensusPipe(consensusPipeConnector)));
+ consensusPipe2ConnectorMap
+ .keySet()
+ .forEach(
+ consensusPipeName ->
+ syncLag =
+ Math.max(syncLag,
getSyncLagForSpecificConsensusPipe(consensusPipeName)));
return syncLag;
} finally {
lock.unlock();
}
}
+ public void clear() {
+ this.consensusPipe2ConnectorMap.clear();
+ }
+
private PipeConsensusSyncLagManager() {
// do nothing
}
@@ -110,6 +148,7 @@ public class PipeConsensusSyncLagManager {
}
public static void release(String groupId) {
+ PipeConsensusSyncLagManager.getInstance(groupId).clear();
PipeConsensusSyncLagManagerHolder.CONSENSU_GROUP_ID_2_INSTANCE_MAP.remove(groupId);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
index 0994935a6d7..42d0830e711 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
@@ -40,7 +40,7 @@ import
org.apache.iotdb.db.consensus.statemachine.dataregion.IoTConsensusDataReg
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.consensus.ConsensusPipeDataNodeDispatcher;
import
org.apache.iotdb.db.pipe.consensus.ConsensusPipeDataNodeRuntimeAgentGuardian;
-import org.apache.iotdb.db.pipe.consensus.ProgressIndexDataNodeManager;
+import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
@@ -170,7 +170,7 @@ public class DataRegionConsensusImpl {
.setConsensusPipeSelector(
() ->
PipeDataNodeAgent.task().getAllConsensusPipe())
.setConsensusPipeReceiver(PipeDataNodeAgent.receiver().pipeConsensus())
- .setProgressIndexManager(new
ProgressIndexDataNodeManager())
+ .setProgressIndexManager(new
ReplicateProgressDataNodeManager())
.setConsensusPipeGuardJobIntervalInSeconds(300)
.build())
.setReplicateMode(ReplicateMode.fromValue(CONF.getIotConsensusV2Mode()))
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
index 3573440472a..a44c30fca63 100644
---
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
@@ -33,6 +33,7 @@ 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;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeConnector;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName;
import org.apache.iotdb.consensus.pipe.metric.PipeConsensusSyncLagManager;
import org.apache.iotdb.consensus.pipe.thrift.TCommitId;
import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq;
@@ -135,7 +136,7 @@ public class PipeConsensusAsyncConnector extends
IoTDBConnector implements Conse
// initialize metric components
pipeConsensusConnectorMetrics = new PipeConsensusConnectorMetrics(this);
PipeConsensusSyncLagManager.getInstance(getConsensusGroupIdStr())
- .addConsensusPipeConnector(this);
+ .addConsensusPipeConnector(new ConsensusPipeName(consensusPipeName),
this);
MetricService.getInstance().addMetricSet(this.pipeConsensusConnectorMetrics);
// In PipeConsensus, one pipeConsensusTask corresponds to a
pipeConsensusConnector. Thus,
@@ -169,7 +170,7 @@ public class PipeConsensusAsyncConnector extends
IoTDBConnector implements Conse
LOGGER.debug(
"PipeConsensus-ConsensusGroup-{}: no.{} event-{} added to
connector buffer",
consensusGroupId,
- event.getCommitId(),
+ event.getReplicateIndexForIoTV2(),
event);
}
// Special judge to avoid transfer stuck when re-transfer events that
will not be put in
@@ -227,7 +228,8 @@ public class PipeConsensusAsyncConnector extends
IoTDBConnector implements Conse
}
iterator.remove();
// update replicate progress
- currentReplicateProgress = Math.max(currentReplicateProgress,
event.getCommitId());
+ currentReplicateProgress =
+ Math.max(currentReplicateProgress, event.getReplicateIndexForIoTV2());
// decrease reference count
event.decreaseReferenceCount(PipeConsensusAsyncConnector.class.getName(),
true);
}
@@ -273,8 +275,7 @@ public class PipeConsensusAsyncConnector extends
IoTDBConnector implements Conse
(PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent;
tCommitId =
new TCommitId(
- pipeInsertNodeTabletInsertionEvent.getCommitId(),
-
pipeInsertNodeTabletInsertionEvent.getCommitterKey().getRestartTimes(),
+ pipeInsertNodeTabletInsertionEvent.getReplicateIndexForIoTV2(),
pipeInsertNodeTabletInsertionEvent.getRebootTimes());
// We increase the reference count for this event to determine if the
event may be released.
@@ -352,8 +353,7 @@ public class PipeConsensusAsyncConnector extends
IoTDBConnector implements Conse
(PipeTsFileInsertionEvent) tsFileInsertionEvent;
TCommitId tCommitId =
new TCommitId(
- pipeTsFileInsertionEvent.getCommitId(),
- pipeTsFileInsertionEvent.getCommitterKey().getRestartTimes(),
+ pipeTsFileInsertionEvent.getReplicateIndexForIoTV2(),
pipeTsFileInsertionEvent.getRebootTimes());
TConsensusGroupId tConsensusGroupId =
new TConsensusGroupId(TConsensusGroupType.DataRegion,
consensusGroupId);
@@ -588,7 +588,7 @@ public class PipeConsensusAsyncConnector extends
IoTDBConnector implements Conse
}
PipeConsensusSyncLagManager.getInstance(getConsensusGroupIdStr())
- .removeConsensusPipeConnector(this);
+ .removeConsensusPipeConnector(new
ConsensusPipeName(consensusPipeName));
MetricService.getInstance().removeMetricSet(this.pipeConsensusConnectorMetrics);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
index 01d3fbabcd1..c92f99fb3ac 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
@@ -247,8 +247,7 @@ public class PipeConsensusSyncConnector extends
IoTDBConnector {
final TPipeConsensusTransferResp resp;
final TCommitId tCommitId =
new TCommitId(
- pipeDeleteDataNodeEvent.getCommitId(),
- pipeDeleteDataNodeEvent.getCommitterKey().getRestartTimes(),
+ pipeDeleteDataNodeEvent.getReplicateIndexForIoTV2(),
pipeDeleteDataNodeEvent.getRebootTimes());
final TConsensusGroupId tConsensusGroupId =
new TConsensusGroupId(TConsensusGroupType.DataRegion,
consensusGroupId);
@@ -317,8 +316,7 @@ public class PipeConsensusSyncConnector extends
IoTDBConnector {
final TPipeConsensusTransferResp resp;
final TCommitId tCommitId =
new TCommitId(
- pipeInsertNodeTabletInsertionEvent.getCommitId(),
-
pipeInsertNodeTabletInsertionEvent.getCommitterKey().getRestartTimes(),
+ pipeInsertNodeTabletInsertionEvent.getReplicateIndexForIoTV2(),
pipeInsertNodeTabletInsertionEvent.getRebootTimes());
final TConsensusGroupId tConsensusGroupId =
new TConsensusGroupId(TConsensusGroupType.DataRegion,
consensusGroupId);
@@ -377,8 +375,7 @@ public class PipeConsensusSyncConnector extends
IoTDBConnector {
syncRetryClientManager.borrowClient(getFollowerUrl())) {
final TCommitId tCommitId =
new TCommitId(
- pipeTsFileInsertionEvent.getCommitId(),
- pipeTsFileInsertionEvent.getCommitterKey().getRestartTimes(),
+ pipeTsFileInsertionEvent.getReplicateIndexForIoTV2(),
pipeTsFileInsertionEvent.getRebootTimes());
final TConsensusGroupId tConsensusGroupId =
new TConsensusGroupId(TConsensusGroupType.DataRegion,
consensusGroupId);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTabletInsertionEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTabletInsertionEventHandler.java
index 347cd67d8c6..609e7e5e509 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTabletInsertionEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTabletInsertionEventHandler.java
@@ -92,8 +92,8 @@ public abstract class
PipeConsensusTabletInsertionEventHandler<E extends TPipeCo
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
LOGGER.info(
- "Debug only: no.{} event successfully processed!",
- ((EnrichedEvent) event).getCommitId());
+ "InsertNodeTransfer: no.{} event successfully processed!",
+ ((EnrichedEvent) event).getReplicateIndexForIoTV2());
connector.removeEventFromBuffer((EnrichedEvent) event);
}
@@ -107,12 +107,12 @@ public abstract class
PipeConsensusTabletInsertionEventHandler<E extends TPipeCo
@Override
public void onError(Exception exception) {
LOGGER.warn(
- "Failed to transfer TabletInsertionEvent {} (committer key={}, commit
id={}).",
+ "Failed to transfer TabletInsertionEvent {} (committer key={},
replicate index={}).",
event instanceof EnrichedEvent
? ((EnrichedEvent) event).coreReportMessage()
: event.toString(),
event instanceof EnrichedEvent ? ((EnrichedEvent)
event).getCommitterKey() : null,
- event instanceof EnrichedEvent ? ((EnrichedEvent) event).getCommitId()
: null,
+ event instanceof EnrichedEvent ? ((EnrichedEvent)
event).getReplicateIndexForIoTV2() : null,
exception);
connector.addFailureEventToRetryQueue(event);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTsFileInsertionEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTsFileInsertionEventHandler.java
index 95e48c6b74a..5839e00b7eb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTsFileInsertionEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTsFileInsertionEventHandler.java
@@ -222,10 +222,10 @@ public class PipeConsensusTsFileInsertionEventHandler
PipeConsensusTsFileInsertionEventHandler.class.getName(), true);
LOGGER.info(
- "Successfully transferred file {} (committer key={}, commit
id={}).",
+ "Successfully transferred file {} (committer key={}, replicate
index={}).",
tsFile,
event.getCommitterKey(),
- event.getCommitId());
+ event.getReplicateIndexForIoTV2());
if (client != null) {
client.setShouldReturnSelf(true);
@@ -274,10 +274,10 @@ public class PipeConsensusTsFileInsertionEventHandler
@Override
public void onError(final Exception exception) {
LOGGER.warn(
- "Failed to transfer TsFileInsertionEvent {} (committer key {}, commit
id {}).",
+ "Failed to transfer TsFileInsertionEvent {} (committer key {},
replicate index {}).",
tsFile,
event.getCommitterKey(),
- event.getCommitId(),
+ event.getReplicateIndexForIoTV2(),
exception);
try {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/payload/builder/PipeConsensusTransferBatchReqBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/payload/builder/PipeConsensusTransferBatchReqBuilder.java
index 7cc3d22388b..e86193dad17 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/payload/builder/PipeConsensusTransferBatchReqBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/payload/builder/PipeConsensusTransferBatchReqBuilder.java
@@ -123,7 +123,7 @@ public abstract class PipeConsensusTransferBatchReqBuilder
implements AutoClosea
return false;
}
- final long requestCommitId = ((EnrichedEvent) event).getCommitId();
+ final long requestCommitId = ((EnrichedEvent)
event).getReplicateIndexForIoTV2();
// The deduplication logic here is to avoid the accumulation of the same
event in a batch when
// retrying.
@@ -182,8 +182,7 @@ public abstract class PipeConsensusTransferBatchReqBuilder
implements AutoClosea
(PipeInsertNodeTabletInsertionEvent) event;
commitId =
new TCommitId(
- pipeInsertNodeTabletInsertionEvent.getCommitId(),
-
pipeInsertNodeTabletInsertionEvent.getCommitterKey().getRestartTimes(),
+ pipeInsertNodeTabletInsertionEvent.getReplicateIndexForIoTV2(),
pipeInsertNodeTabletInsertionEvent.getRebootTimes());
// Read the bytebuffer from the wal file and transfer it directly without
serializing or
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ProgressIndexDataNodeManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ReplicateProgressDataNodeManager.java
similarity index 78%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ProgressIndexDataNodeManager.java
rename to
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ReplicateProgressDataNodeManager.java
index e9ae4ff4454..ce12e841cd9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ProgressIndexDataNodeManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ReplicateProgressDataNodeManager.java
@@ -27,7 +27,8 @@ import
org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName;
-import org.apache.iotdb.consensus.pipe.consensuspipe.ProgressIndexManager;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ReplicateProgressManager;
+import org.apache.iotdb.consensus.pipe.metric.PipeConsensusSyncLagManager;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.storageengine.StorageEngine;
@@ -35,21 +36,32 @@ import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
-public class ProgressIndexDataNodeManager implements ProgressIndexManager {
+public class ReplicateProgressDataNodeManager implements
ReplicateProgressManager {
private static final int DATA_NODE_ID =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+ private static final Map<String, AtomicLong> groupId2ReplicateIndex = new
HashMap<>();
private final Map<ConsensusGroupId, ProgressIndex> groupId2MaxProgressIndex;
+ private final Map<ConsensusPipeName, Long>
consensusPipe2pinnedCommitIndexForMigration;
- public ProgressIndexDataNodeManager() {
+ public ReplicateProgressDataNodeManager() {
this.groupId2MaxProgressIndex = new ConcurrentHashMap<>();
+ this.consensusPipe2pinnedCommitIndexForMigration = new
ConcurrentHashMap<>();
recoverMaxProgressIndexFromDataRegion();
}
+ public static long assignReplicateIndexForIoTV2(String groupId) {
+ return groupId2ReplicateIndex
+ .compute(groupId, (k, v) -> v == null ? new AtomicLong(0) : v)
+ .incrementAndGet();
+ }
+
public static ProgressIndex extractLocalSimpleProgressIndex(ProgressIndex
progressIndex) {
if (progressIndex instanceof RecoverProgressIndex) {
final Map<Integer, SimpleProgressIndex> dataNodeId2LocalIndex =
@@ -129,4 +141,22 @@ public class ProgressIndexDataNodeManager implements
ProgressIndexManager {
public ProgressIndex getMaxAssignedProgressIndex(ConsensusGroupId
consensusGroupId) {
return groupId2MaxProgressIndex.getOrDefault(consensusGroupId,
MinimumProgressIndex.INSTANCE);
}
+
+ @Override
+ public long getSyncLagForSpecificConsensusPipe(
+ ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName) {
+ return PipeConsensusSyncLagManager.getInstance(consensusGroupId.toString())
+ .getSyncLagForRegionMigration(
+ consensusPipeName,
+
this.consensusPipe2pinnedCommitIndexForMigration.getOrDefault(consensusPipeName,
0L));
+ }
+
+ @Override
+ public void pinCommitIndexForMigration(
+ ConsensusGroupId consensusGroupId, ConsensusPipeName consensusPipeName) {
+ this.consensusPipe2pinnedCommitIndexForMigration.put(
+ consensusPipeName,
+ PipeConsensusSyncLagManager.getInstance(consensusGroupId.toString())
+ .getCurrentCommitIndex(consensusPipeName));
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
index 0f6ff3c6315..7e7dec38c66 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.consensus.pipe.PipeConsensus;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
-import org.apache.iotdb.db.pipe.consensus.ProgressIndexDataNodeManager;
+import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
import org.apache.iotdb.db.pipe.consensus.deletion.persist.DeletionBuffer;
import
org.apache.iotdb.db.pipe.consensus.deletion.persist.PageCacheDeletionBuffer;
import org.apache.iotdb.db.pipe.consensus.deletion.recover.DeletionReader;
@@ -176,7 +176,7 @@ public class DeletionResourceManager implements
AutoCloseable {
deleteNode2ResourcesMap.remove(deletionResource.getDeleteDataNode());
// Clean disk
ProgressIndex currentProgressIndex =
- ProgressIndexDataNodeManager.extractLocalSimpleProgressIndex(
+ ReplicateProgressDataNodeManager.extractLocalSimpleProgressIndex(
deletionResource.getProgressIndex());
try (Stream<Path> pathStream = Files.walk(Paths.get(storageDir.getPath()),
1)) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
index 289bcfc53fa..f9f6965f40d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
@@ -25,7 +25,7 @@ import
org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.pipe.consensus.ProgressIndexDataNodeManager;
+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.utils.MmapUtil;
@@ -222,7 +222,8 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
try {
// PipeConsensus ensures that deleteDataNodes use recoverProgressIndex.
ProgressIndex curProgressIndex =
-
ProgressIndexDataNodeManager.extractLocalSimpleProgressIndex(maxProgressIndexInLastFile);
+ ReplicateProgressDataNodeManager.extractLocalSimpleProgressIndex(
+ maxProgressIndexInLastFile);
if (!(curProgressIndex instanceof SimpleProgressIndex)) {
throw new IOException("Invalid deletion progress index: " +
curProgressIndex);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 70f66394dd5..6a94f81017f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -165,7 +165,6 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
synchronized (isClosed) {
isClosed.set(true);
isClosed.notifyAll();
-
// Update flushPointCount after TsFile is closed
flushPointCount = processor.getMemTableFlushPointCount();
}
@@ -351,6 +350,22 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
+ /**
+ * Get ProgressIndex without waiting for tsfile close. Can be used in
getting progressIndex when
+ * memTable becomes immutable.
+ */
+ public ProgressIndex forceGetProgressIndex() {
+ if (resource.isEmpty()) {
+ LOGGER.warn(
+ "Skipping temporary TsFile {}'s progressIndex, will report
MinimumProgressIndex", tsFile);
+ return MinimumProgressIndex.INSTANCE;
+ }
+ if (Objects.nonNull(overridingProgressIndex)) {
+ return overridingProgressIndex;
+ }
+ return resource.getMaxProgressIndex();
+ }
+
@Override
protected void reportProgress() {
super.reportProgress();
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
index ab9c2a51036..200ed98f0ed 100644
---
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,39 +20,62 @@
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.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
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,
final boolean isLoaded,
final boolean isGeneratedByPipe) {
- return TS_FILE_EPOCH_MANAGER.bindPipeTsFileInsertionEvent(
+ PipeTsFileInsertionEvent tsFileInsertionEvent =
new PipeTsFileInsertionEvent(
- isTableModel, databaseNameFromDataRegion, resource, isLoaded,
isGeneratedByPipe, false),
- resource);
+ isTableModel, databaseNameFromDataRegion, resource, isLoaded,
isGeneratedByPipe, false);
+
+ // if using IoTV2, assign a replicateIndex for this event
+ if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
+ && PipeConsensusProcessor.isShouldReplicate(tsFileInsertionEvent)) {
+ tsFileInsertionEvent.setReplicateIndexForIoTV2(
+
ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
+ LOGGER.debug(
+ "[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,
final InsertNode insertNode,
final TsFileResource resource) {
- return TS_FILE_EPOCH_MANAGER.bindPipeInsertNodeTabletInsertionEvent(
+ PipeInsertNodeTabletInsertionEvent insertionEvent =
new PipeInsertNodeTabletInsertionEvent(
isTableModel,
databaseNameFromDataRegion,
@@ -60,9 +83,22 @@ public class PipeRealtimeEventFactory {
insertNode.getTargetPath(),
insertNode.getProgressIndex(),
insertNode.isAligned(),
- insertNode.isGeneratedByPipe()),
- insertNode,
- resource);
+ 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.debug(
+ "[Region{}]Set {} for event {}",
+ dataRegionId,
+ insertionEvent.getReplicateIndexForIoTV2(),
+ insertionEvent);
+ }
+
+ return TS_FILE_EPOCH_MANAGER.bindPipeInsertNodeTabletInsertionEvent(
+ insertionEvent, insertNode, resource);
}
public static PipeRealtimeEvent createRealtimeEvent(
@@ -71,9 +107,24 @@ public class PipeRealtimeEventFactory {
new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null,
null, null);
}
- public static PipeRealtimeEvent createRealtimeEvent(final
AbstractDeleteDataNode node) {
- return new PipeRealtimeEvent(
- new PipeDeleteDataNodeEvent(node, node.isGeneratedByPipe()), null,
null, null, null);
+ public static PipeRealtimeEvent createRealtimeEvent(
+ final String dataRegionId, 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.debug(
+ "[Region{}]Set {} for event {}",
+ dataRegionId,
+ deleteDataNodeEvent.getReplicateIndexForIoTV2(),
+ deleteDataNodeEvent);
+ }
+
+ return new PipeRealtimeEvent(deleteDataNodeEvent, null, null, null, null);
}
public static PipeRealtimeEvent createRealtimeEvent(final
ProgressReportEvent event) {
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
index 74c8aa0a1eb..d8eb3ac6b89 100644
---
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
@@ -32,12 +32,16 @@ import
org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource;
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;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.extractor.dataregion.DataRegionListeningFilter;
+import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
import org.apache.iotdb.db.storageengine.StorageEngine;
@@ -795,6 +799,19 @@ public class
PipeHistoricalDataRegionTsFileAndDeletionExtractor
historicalDataExtractionStartTime,
historicalDataExtractionEndTime);
+ // if using IoTV2, assign a replicateIndex for this event
+ if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
+ && PipeConsensusProcessor.isShouldReplicate(event)) {
+ event.setReplicateIndexForIoTV2(
+ ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(
+ resource.getDataRegionId()));
+ LOGGER.debug(
+ "[Region{}]Set {} for event {}",
+ resource.getDataRegionId(),
+ event.getReplicateIndexForIoTV2(),
+ event);
+ }
+
if (sloppyPattern || isDbNameCoveredByPattern) {
event.skipParsingPattern();
}
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
index 7ef2b1c4841..0950718b097 100644
---
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
@@ -187,6 +187,8 @@ public class PipeDataRegionAssigner implements Closeable {
extractor.getRealtimeDataExtractionStartTime(),
extractor.getRealtimeDataExtractionEndTime());
final EnrichedEvent innerEvent = copiedEvent.getEvent();
+ // Bind replicateIndex for IoTV2
+
innerEvent.setReplicateIndexForIoTV2(event.getEvent().getReplicateIndexForIoTV2());
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
index 138f7c8f0f6..13498d8ca5f 100644
---
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
@@ -115,7 +115,12 @@ public class PipeInsertionDataNodeListener {
assigner.publishToAssign(
PipeRealtimeEventFactory.createRealtimeEvent(
- assigner.isTableModel(), databaseName, tsFileResource, isLoaded,
isGeneratedByPipe));
+ dataRegionId,
+ assigner.isTableModel(),
+ databaseName,
+ tsFileResource,
+ isLoaded,
+ isGeneratedByPipe));
}
public void listenToInsertNode(
@@ -137,7 +142,12 @@ public class PipeInsertionDataNodeListener {
assigner.publishToAssign(
PipeRealtimeEventFactory.createRealtimeEvent(
- assigner.isTableModel(), databaseName, walEntryHandler,
insertNode, tsFileResource));
+ dataRegionId,
+ assigner.isTableModel(),
+ databaseName,
+ walEntryHandler,
+ insertNode,
+ tsFileResource));
}
public DeletionResource listenToDeleteData(
@@ -163,7 +173,7 @@ public class PipeInsertionDataNodeListener {
deletionResource = null;
}
-
assigner.publishToAssign(PipeRealtimeEventFactory.createRealtimeEvent(node));
+
assigner.publishToAssign(PipeRealtimeEventFactory.createRealtimeEvent(regionId,
node));
return deletionResource;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
index c34033aa03c..065b2924846 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
@@ -51,8 +51,13 @@ public class PipeConsensusProcessor implements PipeProcessor
{
public void customize(PipeParameters parameters,
PipeProcessorRuntimeConfiguration configuration)
throws Exception {}
- private boolean isContainLocalData(EnrichedEvent enrichedEvent) {
- final ProgressIndex progressIndex = enrichedEvent.getProgressIndex();
+ private static boolean isContainLocalData(EnrichedEvent enrichedEvent) {
+ ProgressIndex progressIndex;
+ if (enrichedEvent instanceof PipeTsFileInsertionEvent) {
+ progressIndex = ((PipeTsFileInsertionEvent)
enrichedEvent).forceGetProgressIndex();
+ } else {
+ progressIndex = enrichedEvent.getProgressIndex();
+ }
if (progressIndex instanceof RecoverProgressIndex) {
return ((RecoverProgressIndex) progressIndex)
.getDataNodeId2LocalIndex()
@@ -71,27 +76,41 @@ public class PipeConsensusProcessor implements
PipeProcessor {
return false;
}
+ /**
+ * Judge whether a given event should be replicated or sent by consensus
pipe. Only events that
+ * pass this check can be assigned a specific value for
`EnrichedEvent.replicateIndexForIoTV2`
+ *
+ * @param enrichedEvent
+ * @return true if event is totally generated by client operation instead of
consensus
+ * replication.
+ */
+ public static boolean isShouldReplicate(EnrichedEvent enrichedEvent) {
+ if (enrichedEvent instanceof TsFileInsertionEvent
+ && ((PipeTsFileInsertionEvent)
enrichedEvent).isGeneratedByPipeConsensus()) {
+ // Only user-generated TsFileInsertionEvent can be replicated. Any
tsFile synchronized from a
+ // replica should not be replicated again
+ return false;
+ }
+ // Only user-generated TabletInsertionEvent can be replicated.
+ return isContainLocalData(enrichedEvent);
+ }
+
@Override
public void process(TsFileInsertionEvent tsFileInsertionEvent,
EventCollector eventCollector)
throws Exception {
- // Only user-generated TsFileInsertionEvent can be replicated. Any tsFile
synchronized from a
- // replica should not be replicated again
if (tsFileInsertionEvent instanceof EnrichedEvent
- && !((PipeTsFileInsertionEvent)
tsFileInsertionEvent).isGeneratedByPipeConsensus()) {
- final EnrichedEvent enrichedEvent = (EnrichedEvent) tsFileInsertionEvent;
- if (isContainLocalData(enrichedEvent)) {
- eventCollector.collect(tsFileInsertionEvent);
- }
+ && ((EnrichedEvent) tsFileInsertionEvent).getReplicateIndexForIoTV2()
+ != EnrichedEvent.NO_COMMIT_ID) {
+ eventCollector.collect(tsFileInsertionEvent);
}
}
@Override
public void process(TabletInsertionEvent tabletInsertionEvent,
EventCollector eventCollector)
throws Exception {
- // Only user-generated TabletInsertionEvent can be replicated.
if (tabletInsertionEvent instanceof EnrichedEvent) {
final EnrichedEvent enrichedEvent = (EnrichedEvent) tabletInsertionEvent;
- if (isContainLocalData(enrichedEvent)) {
+ if (enrichedEvent.getReplicateIndexForIoTV2() !=
EnrichedEvent.NO_COMMIT_ID) {
eventCollector.collect(tabletInsertionEvent);
}
}
@@ -101,7 +120,7 @@ public class PipeConsensusProcessor implements
PipeProcessor {
public void process(Event event, EventCollector eventCollector) throws
Exception {
if (event instanceof EnrichedEvent) {
final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
- if (isContainLocalData(enrichedEvent)) {
+ if (enrichedEvent.getReplicateIndexForIoTV2() !=
EnrichedEvent.NO_COMMIT_ID) {
eventCollector.collect(event);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
index 0a5ba1e9782..f974926c5ad 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
@@ -975,7 +975,7 @@ public class PipeConsensusReceiver {
consensusPipeName.toString());
}
- // initiate receiverFileDirWithIdSuffix
+ // initiate receiverFileBaseDir
String receiverFileBaseDir;
try {
receiverFileBaseDir = getReceiverFileBaseDir();
@@ -1321,17 +1321,15 @@ public class PipeConsensusReceiver {
private final PipeConsensusTsFileWriterPool tsFileWriterPool;
private final AtomicInteger WALEventCount = new AtomicInteger(0);
private final AtomicInteger tsFileEventCount = new AtomicInteger(0);
- private long onSyncedCommitIndex = 0;
+ private long onSyncedReplicateIndex = 0;
private int connectorRebootTimes = 0;
- private int pipeTaskRestartTimes = 0;
public RequestExecutor(
PipeConsensusReceiverMetrics metric, PipeConsensusTsFileWriterPool
tsFileWriterPool) {
this.reqExecutionOrderBuffer =
new TreeSet<>(
Comparator.comparingInt(RequestMeta::getDataNodeRebootTimes)
- .thenComparingInt(RequestMeta::getPipeTaskRestartTimes)
- .thenComparingLong(RequestMeta::getCommitIndex));
+ .thenComparingLong(RequestMeta::getReplicateIndex));
this.lock = new ReentrantLock();
this.condition = lock.newCondition();
this.metric = metric;
@@ -1344,7 +1342,7 @@ public class PipeConsensusReceiver {
consensusPipeName,
commitId);
RequestMeta curMeta = reqExecutionOrderBuffer.pollFirst();
- onSyncedCommitIndex = commitId.getCommitIndex();
+ onSyncedReplicateIndex = commitId.getReplicateIndex();
// update metric, notice that curMeta is never null.
if (isTransferTsFileSeal) {
tsFileEventCount.decrementAndGet();
@@ -1391,28 +1389,11 @@ public class PipeConsensusReceiver {
consensusPipeName);
return new TPipeConsensusTransferResp(status);
}
- // Similarly, check pipeTask restartTimes
- if (tCommitId.getDataNodeRebootTimes() == connectorRebootTimes
- && tCommitId.getPipeTaskRestartTimes() < pipeTaskRestartTimes) {
- final TSStatus status =
- new TSStatus(
- RpcUtils.getStatus(
- TSStatusCode.PIPE_CONSENSUS_DEPRECATED_REQUEST,
- "PipeConsensus receiver received a deprecated request,
which may be sent before the pipe task restart. Consider to discard it"));
- LOGGER.info(
- "PipeConsensus-PipeName-{}: received a deprecated request, which
may be sent before the pipe task restart. Consider to discard it",
- consensusPipeName);
- return new TPipeConsensusTransferResp(status);
- }
// Judge whether connector has rebooted or not, if the rebootTimes
increases compared to
// connectorRebootTimes, need to reset receiver because connector has
been restarted.
if (tCommitId.getDataNodeRebootTimes() > connectorRebootTimes) {
resetWithNewestRebootTime(tCommitId.getDataNodeRebootTimes(),
condition);
}
- // Similarly, check pipeTask restartTimes
- if (tCommitId.getPipeTaskRestartTimes() > pipeTaskRestartTimes) {
- resetWithNewestRestartTime(tCommitId.getPipeTaskRestartTimes(),
condition);
- }
// update metric
if (isTransferTsFilePiece &&
!reqExecutionOrderBuffer.contains(requestMeta)) {
// only update tsFileEventCount when tsFileEvent is first enqueue.
@@ -1444,7 +1425,7 @@ public class PipeConsensusReceiver {
// Polling to process
while (true) {
if (reqExecutionOrderBuffer.first().equals(requestMeta)
- && tCommitId.getCommitIndex() == onSyncedCommitIndex + 1) {
+ && tCommitId.getReplicateIndex() == onSyncedReplicateIndex + 1) {
long startApplyNanos = System.nanoTime();
metric.recordDispatchWaitingTimer(startApplyNanos -
startDispatchNanos);
requestMeta.setStartApplyNanos(startApplyNanos);
@@ -1529,7 +1510,7 @@ public class PipeConsensusReceiver {
LOGGER.warn(
"PipeConsensus-PipeName-{}: current waiting is interrupted.
onSyncedCommitIndex: {}. Exception: ",
consensusPipeName,
- tCommitId.getCommitIndex(),
+ tCommitId.getReplicateIndex(),
e);
// Avoid infinite loop when RPC thread is killed by OS
return new TPipeConsensusTransferResp(
@@ -1559,26 +1540,12 @@ public class PipeConsensusReceiver {
condition.signalAll();
// sync the follower's connectorRebootTimes with connector's actual
rebootTimes.
this.connectorRebootTimes = connectorRebootTimes;
- // Note: dataNode rebooting will reset pipeTaskRestartTimes.
- this.pipeTaskRestartTimes = 0;
- }
-
- private void resetWithNewestRestartTime(int pipeTaskRestartTimes,
Condition condition) {
- LOGGER.info(
- "PipeConsensus-PipeName-{}: receiver detected an newer
pipeTaskRestartTimes, which indicates the pipe task has restarted. receiver
will reset all its data.",
- consensusPipeName);
- // since pipe task will resend all data that hasn't synchronized after
restarts, it's safe to
- // clear all events in buffer.
- clear();
- // signal all deprecated requests that may wait on condition to expire
them
- condition.signalAll();
- this.pipeTaskRestartTimes = pipeTaskRestartTimes;
}
private void clear() {
this.reqExecutionOrderBuffer.clear();
this.tsFileWriterPool.handleExit(consensusPipeName);
- this.onSyncedCommitIndex = 0;
+ this.onSyncedReplicateIndex = 0;
}
}
@@ -1594,12 +1561,8 @@ public class PipeConsensusReceiver {
return commitId.getDataNodeRebootTimes();
}
- public int getPipeTaskRestartTimes() {
- return commitId.getPipeTaskRestartTimes();
- }
-
- public long getCommitIndex() {
- return commitId.getCommitIndex();
+ public long getReplicateIndex() {
+ return commitId.getReplicateIndex();
}
public void setStartApplyNanos(long startApplyNanos) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
index 30c1d1dca6b..98f90901942 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
@@ -61,6 +61,8 @@ public abstract class EnrichedEvent implements Event {
protected CommitterKey committerKey;
public static final long NO_COMMIT_ID = -1;
protected long commitId = NO_COMMIT_ID;
+ // Used in IoTConsensusV2
+ protected long replicateIndexForIoTV2 = NO_COMMIT_ID;
protected int rebootTimes = 0;
protected final TreePattern treePattern;
@@ -415,6 +417,14 @@ public abstract class EnrichedEvent implements Event {
return commitId;
}
+ public long getReplicateIndexForIoTV2() {
+ return replicateIndexForIoTV2;
+ }
+
+ public void setReplicateIndexForIoTV2(long replicateIndexForIoTV2) {
+ this.replicateIndexForIoTV2 = replicateIndexForIoTV2;
+ }
+
public void onCommitted() {
onCommittedHooks.forEach(Supplier::get);
}
@@ -441,7 +451,8 @@ public abstract class EnrichedEvent implements Event {
final EnrichedEvent otherEvent = (EnrichedEvent) o;
return Objects.equals(committerKey, otherEvent.committerKey)
&& commitId == otherEvent.commitId
- && rebootTimes == otherEvent.rebootTimes;
+ && rebootTimes == otherEvent.rebootTimes
+ && replicateIndexForIoTV2 == otherEvent.replicateIndexForIoTV2;
}
@Override
@@ -459,6 +470,8 @@ public abstract class EnrichedEvent implements Event {
+ committerKey
+ "', commitId="
+ commitId
+ + "', replicateIndexForIoTV2="
+ + replicateIndexForIoTV2
+ ", treePattern='"
+ treePattern
+ "', tablePattern='"
@@ -488,6 +501,8 @@ public abstract class EnrichedEvent implements Event {
+ committerKey
+ "', commitId="
+ commitId
+ + "', replicateIndexForIoTV2="
+ + replicateIndexForIoTV2
+ ", treePattern='"
+ treePattern
+ "', tablePattern='"
diff --git
a/iotdb-protocol/thrift-consensus/src/main/thrift/pipeconsensus.thrift
b/iotdb-protocol/thrift-consensus/src/main/thrift/pipeconsensus.thrift
index ca7a043c007..9bae689e692 100644
--- a/iotdb-protocol/thrift-consensus/src/main/thrift/pipeconsensus.thrift
+++ b/iotdb-protocol/thrift-consensus/src/main/thrift/pipeconsensus.thrift
@@ -21,9 +21,8 @@ include "common.thrift"
namespace java org.apache.iotdb.consensus.pipe.thrift
struct TCommitId {
- 1:required i64 commitIndex
- 2:required i32 pipeTaskRestartTimes
- 3:required i32 dataNodeRebootTimes
+ 1:required i64 replicateIndex
+ 2:required i32 dataNodeRebootTimes
}
struct TPipeConsensusTransferReq {
