You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2023/12/25 10:26:50 UTC
(iotdb) branch master updated: DataRegion should return WRITE_PROCESS_REJECT to Consensus when insert rejected (#11782)
This is an automated email from the ASF dual-hosted git repository.
haonan 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 71a03063427 DataRegion should return WRITE_PROCESS_REJECT to Consensus when insert rejected (#11782)
71a03063427 is described below
commit 71a0306342767887dd5625f8237c4c5f1a5d3698
Author: Haonan <hh...@outlook.com>
AuthorDate: Mon Dec 25 18:26:44 2023 +0800
DataRegion should return WRITE_PROCESS_REJECT to Consensus when insert rejected (#11782)
---
.../consensus/iot/client/DispatchLogHandler.java | 3 +-
.../dataregion/DataExecutionVisitor.java | 35 +++++++++++++++++++++-
.../dataregion/DataRegionStateMachine.java | 3 +-
.../db/storageengine/dataregion/DataRegion.java | 4 +--
.../dataregion/memtable/TsFileProcessor.java | 2 +-
5 files changed, 39 insertions(+), 8 deletions(-)
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java
index b0c8e5f35c6..88a8d90ebaa 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java
@@ -84,8 +84,7 @@ public class DispatchLogHandler implements AsyncMethodCallback<TSyncLogEntriesRe
public static boolean needRetry(int statusCode) {
return statusCode == TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()
|| statusCode == TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()
- || statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()
- || statusCode == TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode();
+ || statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java
index 845532bc96d..f1342d90899 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.WriteProcessException;
+import org.apache.iotdb.db.exception.WriteProcessRejectException;
import org.apache.iotdb.db.exception.query.OutOfTTLException;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
@@ -61,6 +62,9 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
} catch (OutOfTTLException e) {
LOGGER.warn("Error in executing plan node: {}, caused by {}", node, e.getMessage());
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } catch (WriteProcessRejectException e) {
+ LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
} catch (WriteProcessException e) {
LOGGER.error("Error in executing plan node: {}", node, e);
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
@@ -75,6 +79,9 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
} catch (OutOfTTLException e) {
LOGGER.warn("Error in executing plan node: {}, caused by {}", node, e.getMessage());
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } catch (WriteProcessRejectException e) {
+ LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
} catch (WriteProcessException e) {
LOGGER.error("Error in executing plan node: {}", node, e);
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
@@ -85,7 +92,18 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
node.getTimes()[0],
node.getMeasurements(),
e.getFailingStatus());
- return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode());
+ // for each error
+ TSStatus firstStatus = null;
+ for (TSStatus status : e.getFailingStatus()) {
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ firstStatus = status;
+ }
+ // return WRITE_PROCESS_REJECT directly for the consensus retry logic
+ if (status.getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
+ return status;
+ }
+ }
+ return firstStatus;
}
}
@@ -109,6 +127,10 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
insertRowNode.getTime(),
insertRowNode.getMeasurements(),
failedEntry.getValue());
+ // return WRITE_PROCESS_REJECT directly for the consensus retry logic
+ if (failedEntry.getValue().getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
+ return failedEntry.getValue();
+ }
}
return firstStatus;
}
@@ -134,6 +156,10 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
insertTabletNode.getTimes()[0],
insertTabletNode.getMeasurements(),
failedEntry.getValue());
+ // return WRITE_PROCESS_REJECT directly for the consensus retry logic
+ if (failedEntry.getValue().getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
+ return failedEntry.getValue();
+ }
}
return firstStatus;
}
@@ -145,6 +171,9 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
try {
dataRegion.insert(node);
return StatusUtils.OK;
+ } catch (WriteProcessRejectException e) {
+ LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
} catch (WriteProcessException e) {
LOGGER.error("Error in executing plan node: {}", node, e);
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
@@ -162,6 +191,10 @@ public class DataExecutionVisitor extends PlanVisitor<TSStatus, DataRegion> {
insertRowNode.getTime(),
insertRowNode.getMeasurements(),
failedEntry.getValue());
+ // return WRITE_PROCESS_REJECT directly for the consensus retry logic
+ if (failedEntry.getValue().getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
+ return failedEntry.getValue();
+ }
}
return firstStatus;
}
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 88c5307ab44..38a973d554d 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
@@ -305,7 +305,6 @@ public class DataRegionStateMachine extends BaseStateMachine {
// To fix the atomicity problem, we only need to add retry for system reject.
// In other cases, such as readonly, we can return directly because there are retries at the
// consensus layer.
- return statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()
- || statusCode == TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode();
+ return statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index bdbe5449115..48d3c29a2d9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -3056,10 +3056,10 @@ public class DataRegion implements IDataRegionForQuery {
InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i);
try {
insertTablet(insertTabletNode);
- } catch (WriteProcessException | BatchProcessException e) {
+ } catch (WriteProcessException e) {
insertMultiTabletsNode
.getResults()
- .put(i, new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode()));
+ .put(i, RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 1c96bf89ef9..861f69fecf4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -707,7 +707,7 @@ public class TsFileProcessor {
private void updateMemoryInfo(
long memTableIncrement, long chunkMetadataIncrement, long textDataIncrement)
- throws WriteProcessException {
+ throws WriteProcessRejectException {
memTableIncrement += textDataIncrement;
dataRegionInfo.addStorageGroupMemCost(memTableIncrement);
tsFileProcessorInfo.addTSPMemCost(chunkMetadataIncrement);