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 3eddf083968 Fix pipe write reject retry handling (#18098)
3eddf083968 is described below
commit 3eddf0839686b0677ef028c572ffcc6f6adeb3c1
Author: Caideyipi <[email protected]>
AuthorDate: Fri Jul 3 12:33:33 2026 +0800
Fix pipe write reject retry handling (#18098)
---
.../dataregion/DataRegionStateMachine.java | 3 +-
.../dataregion/DataRegionStateMachineTest.java | 93 ++++++++++++++++++++++
2 files changed, 95 insertions(+), 1 deletion(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
index b99304055a9..bf31a16dfd8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
@@ -248,7 +248,8 @@ public class DataRegionStateMachine extends
BaseStateMachine {
int retryTime = 0;
while (retryTime < MAX_WRITE_RETRY_TIMES) {
result = planNode.accept(new DataExecutionVisitor(), region);
- if (needRetry(result.getCode())) {
+ // Let pipe retry with the original event instead of retrying a possibly
mutated node here.
+ if (needRetry(result.getCode()) && !planNode.isGeneratedByPipe()) {
retryTime++;
logger.debug(
"write operation failed because {}, retryTime: {}.",
result.getCode(), retryTime);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachineTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachineTest.java
index c2ccb56fa72..c6ab83d7b6f 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachineTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachineTest.java
@@ -19,18 +19,26 @@
package org.apache.iotdb.db.consensus.statemachine.dataregion;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
+import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.IPlanVisitor;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.enums.TSDataType;
import org.junit.Assert;
import org.junit.Test;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
public class DataRegionStateMachineTest {
@@ -119,4 +127,89 @@ public class DataRegionStateMachineTest {
InsertNode mergedNode = list.get(0).mergeInsertNode(list);
Assert.assertTrue(mergedNode instanceof InsertRowsNode);
}
+
+ @Test
+ public void testPipeGeneratedWriteProcessRejectSkipsStateMachineRetry() {
+ final DataRegionStateMachine stateMachine = new
DataRegionStateMachine(null);
+ final RetryControlledPlanNode planNode = new RetryControlledPlanNode(true);
+
+ final TSStatus status = stateMachine.write(planNode);
+
+ Assert.assertEquals(TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(),
status.getCode());
+ Assert.assertEquals(1, planNode.getAcceptCount());
+ }
+
+ @Test
+ public void testNonPipeWriteProcessRejectCanStillRetry() {
+ final DataRegionStateMachine stateMachine = new
DataRegionStateMachine(null);
+ final RetryControlledPlanNode planNode = new
RetryControlledPlanNode(false);
+
+ final TSStatus status = stateMachine.write(planNode);
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(2, planNode.getAcceptCount());
+ }
+
+ private static class RetryControlledPlanNode extends PlanNode {
+
+ private final boolean markAsPipeOnFirstAccept;
+ private int acceptCount = 0;
+
+ private RetryControlledPlanNode(final boolean markAsPipeOnFirstAccept) {
+ super(new PlanNodeId("test"));
+ this.markAsPipeOnFirstAccept = markAsPipeOnFirstAccept;
+ }
+
+ private int getAcceptCount() {
+ return acceptCount;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R, C> R accept(final IPlanVisitor<R, C> visitor, final C context) {
+ ++acceptCount;
+ if (markAsPipeOnFirstAccept) {
+ markAsGeneratedByPipe();
+ }
+ return (R)
+ new TSStatus(
+ (acceptCount == 1 ? TSStatusCode.WRITE_PROCESS_REJECT :
TSStatusCode.SUCCESS_STATUS)
+ .getStatusCode());
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void addChild(final PlanNode child) {
+ // No children.
+ }
+
+ @Override
+ public PlanNode clone() {
+ return new RetryControlledPlanNode(markAsPipeOnFirstAccept);
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return NO_CHILD_ALLOWED;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected void serializeAttributes(final ByteBuffer byteBuffer) {
+ // This test node is never serialized.
+ }
+
+ @Override
+ protected void serializeAttributes(final DataOutputStream stream) throws
IOException {
+ // This test node is never serialized.
+ }
+ }
}
