This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 3fd6e26eb32 fix(consensus): preserve SyncLog retry ordering (#18323)
3fd6e26eb32 is described below
commit 3fd6e26eb32930167e201bde6c67114f2b40f537
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 30 10:12:11 2026 +0800
fix(consensus): preserve SyncLog retry ordering (#18323)
* test(consensus): expose SyncLog retry ordering issue
* fix(consensus): preserve SyncLog retry ordering
---------
Co-authored-by: Tian Jiang <[email protected]>
---
.../iotdb/consensus/i18n/IoTConsensusMessages.java | 3 +
.../iotdb/consensus/i18n/IoTConsensusMessages.java | 3 +
.../consensus/iot/IoTConsensusServerImpl.java | 18 ++++-
.../apache/iotdb/consensus/iot/ReplicateTest.java | 83 ++++++++++++++++++++++
4 files changed, 105 insertions(+), 2 deletions(-)
diff --git
a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
index 24efec2dd60..59e3567109b 100644
---
a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
+++
b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
@@ -259,6 +259,9 @@ public final class IoTConsensusMessages {
public static final String
MESSAGE_SYNC_LOG_REQUEST_WITH_SYNC_INDEX_ARG_WAS_INTERRUPTED_WHILE_WAITING_81B4ABB2
=
"SyncLog request with SyncIndex %d was interrupted while waiting";
+ public static final String
+
MESSAGE_THE_REQUEST_MUST_WAIT_FOR_THE_PREVIOUS_REQUEST_TO_COMPLETE_470849A7 =
+ "The request must wait for the previous request to complete";
// ===================== SyncStatus =====================
diff --git
a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
index 810ae0a3761..5164b57b9f5 100644
---
a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
+++
b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
@@ -257,6 +257,9 @@ public final class IoTConsensusMessages {
public static final String
MESSAGE_SYNC_LOG_REQUEST_WITH_SYNC_INDEX_ARG_WAS_INTERRUPTED_WHILE_WAITING_81B4ABB2
=
"等待期间 SyncLog 请求（SyncIndex %d）被中断";
+ public static final String
+
MESSAGE_THE_REQUEST_MUST_WAIT_FOR_THE_PREVIOUS_REQUEST_TO_COMPLETE_470849A7 =
+ "该请求必须等待前置请求完成";
// ===================== SyncStatus =====================
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 430507a4048..b0bf44decb6 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
@@ -35,6 +35,7 @@ import
org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.consensus.IStateMachine;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.Peer;
@@ -1453,9 +1454,22 @@ public class IoTConsensusServerImpl {
long sortTime = System.nanoTime();
ioTConsensusServerMetrics.recordSortCost(sortTime - insertStartTime);
List<TSStatus> subStatus = new LinkedList<>();
- for (IConsensusRequest insertNode : request.getInsertNodes()) {
+ List<IConsensusRequest> insertNodes = request.getInsertNodes();
+ for (int i = 0; i < insertNodes.size(); i++) {
+ IConsensusRequest insertNode = insertNodes.get(i);
insertNode.markAsGeneratedByRemoteConsensusLeader();
- subStatus.add(stateMachine.write(insertNode));
+ TSStatus status = stateMachine.write(insertNode);
+ subStatus.add(status);
+ if (RetryUtils.needRetryForWrite(status.getCode())) {
+ for (int j = i + 1; j < insertNodes.size(); j++) {
+ subStatus.add(
+ RpcUtils.getStatus(
+ TSStatusCode.WRITE_PROCESS_REJECT,
+ IoTConsensusMessages
+
.MESSAGE_THE_REQUEST_MUST_WAIT_FOR_THE_PREVIOUS_REQUEST_TO_COMPLETE_470849A7));
+ }
+ break;
+ }
}
if (subStatus.stream()
.allMatch(status -> status.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode())) {
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
index e096f559773..fad1fbea334 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.request.IConsensusRequest;
+import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.common.ConsensusGroup;
import org.apache.iotdb.consensus.common.Peer;
@@ -402,6 +403,60 @@ public class ReplicateTest {
Assert.assertTrue(failingStateMachine.getData().contains(entry));
}
+ /**
+ * Verifies that retrying a SyncLog batch after a transient write failure
preserves the original
+ * application order on the follower.
+ */
+ @Test
+ public void syncLogWriteProcessErrorRetryPreservesRequestOrderTest() throws
Exception {
+ IoTConsensusConfig retryConfig =
+ IoTConsensusConfig.newBuilder()
+ .setReplication(
+
IoTConsensusConfig.Replication.newBuilder().setMaxPendingBatchesNum(1).build())
+ .build();
+ servers
+ .get(0)
+ .reloadConsensusConfig(
+
ConsensusConfig.newBuilder().setIoTConsensusConfig(retryConfig).build());
+
+ OrderTrackingWriteProcessErrorOnceTestStateMachine failingStateMachine =
+ new OrderTrackingWriteProcessErrorOnceTestStateMachine();
+ stateMachines.set(0, failingStateMachine);
+ servers.get(0).createLocalPeer(group.getGroupId(), group.getPeers());
+ IoTConsensusServerImpl server = servers.get(0).getImpl(gid);
+
+ DeserializedBatchIndexedConsensusRequest request =
+ new DeserializedBatchIndexedConsensusRequest(1, 2, 2,
peers.get(1).getNodeId(), 2);
+ request.add(
+ new IndexedConsensusRequest(
+ 1, 1, Collections.singletonList(new TestEntry(1, peers.get(1)))));
+ request.add(
+ new IndexedConsensusRequest(
+ 2, 2, Collections.singletonList(new TestEntry(2, peers.get(1)))));
+
+ TSStatus firstAttempt = server.syncLog(peers.get(1).getNodeId(), request);
+ Assert.assertEquals(
+ TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode(),
+ firstAttempt.getSubStatus().get(0).getCode());
+ Assert.assertEquals(
+ TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(),
+ firstAttempt.getSubStatus().get(1).getCode());
+ Assert.assertEquals(
+ IoTConsensusMessages
+
.MESSAGE_THE_REQUEST_MUST_WAIT_FOR_THE_PREVIOUS_REQUEST_TO_COMPLETE_470849A7,
+ firstAttempt.getSubStatus().get(1).getMessage());
+ Assert.assertTrue(
+ firstAttempt.getSubStatus().stream()
+ .anyMatch(status ->
RetryUtils.needRetryForWrite(status.getCode())));
+
+ TSStatus retryAttempt = server.syncLog(peers.get(1).getNodeId(), request);
+ Assert.assertTrue(
+ retryAttempt.getSubStatus().stream()
+ .allMatch(status -> status.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ Assert.assertEquals(
+ Arrays.asList(1L, 2L),
failingStateMachine.getFirstSuccessfullyAppliedSyncIndexes());
+ }
+
@Test
public void parsingAndConstructIDTest() throws Exception {
logger.info("Start ParsingAndConstructIDTest");
@@ -453,4 +508,32 @@ public class ReplicateTest {
return writeAttempts.get();
}
}
+
+ private static class OrderTrackingWriteProcessErrorOnceTestStateMachine
+ extends WriteProcessErrorOnceTestStateMachine {
+
+ private final List<Long> firstSuccessfullyAppliedSyncIndexes =
+ Collections.synchronizedList(new ArrayList<>());
+
+ @Override
+ public TSStatus write(IConsensusRequest request) {
+ TSStatus status = super.write(request);
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && request instanceof IndexedConsensusRequest) {
+ long syncIndex = ((IndexedConsensusRequest) request).getSyncIndex();
+ synchronized (firstSuccessfullyAppliedSyncIndexes) {
+ if (!firstSuccessfullyAppliedSyncIndexes.contains(syncIndex)) {
+ firstSuccessfullyAppliedSyncIndexes.add(syncIndex);
+ }
+ }
+ }
+ return status;
+ }
+
+ private List<Long> getFirstSuccessfullyAppliedSyncIndexes() {
+ synchronized (firstSuccessfullyAppliedSyncIndexes) {
+ return new ArrayList<>(firstSuccessfullyAppliedSyncIndexes);
+ }
+ }
+ }
}
