This is an automated email from the ASF dual-hosted git repository.
JackieTien97 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 2fd1fe3fda4 support the retry for the execution the insert path in
statemachine with DN metadata fenced and retry for coordinator (#18301)
2fd1fe3fda4 is described below
commit 2fd1fe3fda40012c68b51938cdff2405079aec3a
Author: CYB <[email protected]>
AuthorDate: Mon Jul 27 17:25:34 2026 +0800
support the retry for the execution the insert path in statemachine with DN
metadata fenced and retry for coordinator (#18301)
---
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 4 +-
.../ratis/ApplicationStateMachineProxy.java | 32 ++++-
.../dataregion/DataExecutionVisitor.java | 11 ++
.../dataregion/DataRegionStateMachine.java | 14 +-
.../analyze/cache/partition/PartitionCache.java | 4 +-
.../cache/TreeDeviceSchemaCacheManager.java | 4 +-
.../schemaengine/lease/MetadataLeaseManager.java | 6 +-
.../db/schemaengine/table/DataNodeTableCache.java | 59 ++++++---
.../iotdb/db/schemaengine/table/ITableCache.java | 7 +
.../db/storageengine/dataregion/DataRegion.java | 5 +-
.../dataregion/DataRegionStateMachineTest.java | 143 +++++++++++++++++++++
.../schemaengine/lease/MetadataLeaseTestUtils.java | 9 +-
.../table/DataNodeTableCacheLeaseTest.java | 24 +++-
.../exception/MetadataLeaseFencedException.java | 21 ++-
.../org/apache/iotdb/commons/utils/RetryUtils.java | 1 +
.../apache/iotdb/commons/utils/StatusUtils.java | 1 +
.../iotdb/commons/utils/RetryStatusUtilsTest.java} | 27 +++-
17 files changed, 322 insertions(+), 50 deletions(-)
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 384f38028e8..8f28c916056 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -90,7 +90,9 @@ public enum TSStatusCode {
TYPE_NOT_FOUND(528),
DATABASE_CONFLICT(529),
DATABASE_MODEL(530),
- METADATA_LEASE_FENCED(531),
+ // the range [531, 534] has been occupied
+ METADATA_LEASE_FENCED(535),
+ METADATA_LEASE_FENCED_RETRY_REQUIRED(536),
TABLE_NOT_EXISTS(550),
TABLE_ALREADY_EXISTS(551),
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
index d7900cf2388..6c0366c546f 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
@@ -57,6 +57,7 @@ import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
public class ApplicationStateMachineProxy extends BaseStateMachine {
@@ -152,6 +153,10 @@ public class ApplicationStateMachineProxy extends
BaseStateMachine {
deserializedRequest.markAsGeneratedByRemoteConsensusLeader();
}
final TSStatus result =
applicationStateMachine.write(deserializedRequest);
+ if (result.getCode() ==
TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()
+ && waitBeforeRetry()) {
+ continue;
+ }
ret = new ResponseMessage(result);
break;
} catch (Throwable rte) {
@@ -160,13 +165,11 @@ public class ApplicationStateMachineProxy extends
BaseStateMachine {
new ResponseMessage(
new
TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())
.setMessage(RatisMessages.INTERNAL_ERROR_STATEMACHINE_RUNTIME_EXCEPTION + rte));
- if (Utils.stallApply(consensusGroupType)) {
- waitUntilSystemAllowApply();
- } else {
+ if (!Utils.stallApply(consensusGroupType) ||
!waitUntilSystemAllowApply()) {
break;
}
}
- } while (Utils.stallApply(consensusGroupType));
+ } while (true);
if (isLeader) {
// only record time cost for data region in Performance Overview
Dashboard
@@ -182,16 +185,35 @@ public class ApplicationStateMachineProxy extends
BaseStateMachine {
return CompletableFuture.completedFuture(ret);
}
- private void waitUntilSystemAllowApply() {
+ /**
+ * @return true if the wait completed normally, false if interrupted
+ */
+ private boolean waitUntilSystemAllowApply() {
try {
Retriable.attemptUntilTrue(
() -> !Utils.stallApply(consensusGroupType),
TimeDuration.ONE_MINUTE,
"waitUntilSystemAllowApply",
logger);
+ return true;
+ } catch (InterruptedException e) {
+ logger.warn(RatisMessages.INTERRUPTED_WAITING_SYSTEM_READY, this, e);
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ /**
+ * @return true if the sleep completed normally, false if interrupted
+ */
+ private boolean waitBeforeRetry() {
+ try {
+ TimeUnit.MINUTES.sleep(1);
+ return true;
} catch (InterruptedException e) {
logger.warn(RatisMessages.INTERRUPTED_WAITING_SYSTEM_READY, this, e);
Thread.currentThread().interrupt();
+ return false;
}
}
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 bd48ed25ae7..96bbb6ccc16 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
@@ -21,6 +21,7 @@ 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.exception.MetadataLeaseFencedException;
import org.apache.iotdb.commons.exception.SemanticException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
@@ -88,6 +89,8 @@ public class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
} catch (WriteProcessException e) {
LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE, node, e);
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } catch (MetadataLeaseFencedException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
}
}
@@ -132,6 +135,8 @@ public class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
}
}
return firstStatus;
+ } catch (final MetadataLeaseFencedException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
}
}
@@ -170,6 +175,8 @@ public class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
} catch (SemanticException | TableLostRuntimeException e) {
LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED,
node, e.getMessage());
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } catch (MetadataLeaseFencedException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
}
}
@@ -205,6 +212,8 @@ public class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
}
}
return firstStatus;
+ } catch (MetadataLeaseFencedException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
}
}
@@ -243,6 +252,8 @@ public class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
}
}
return firstStatus;
+ } catch (MetadataLeaseFencedException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
}
}
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 8887054e2f2..051dcbb2771 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
@@ -253,7 +253,7 @@ public class DataRegionStateMachine extends
BaseStateMachine {
while (retryTime < MAX_WRITE_RETRY_TIMES) {
result = planNode.accept(new DataExecutionVisitor(), region);
// Let pipe retry with the original event instead of retrying a possibly
mutated node here.
- if (needRetry(result.getCode()) && !planNode.isGeneratedByPipe()) {
+ if (needRetryForSpecificCases(result.getCode(), planNode)) {
retryTime++;
logger.debug(
DataNodePipeMessages.PIPE_LOG_WRITE_OPERATION_FAILED_BECAUSE_RETRYTIME_34EFBE99,
@@ -330,10 +330,16 @@ public class DataRegionStateMachine extends
BaseStateMachine {
}
}
- public static boolean needRetry(int statusCode) {
- // To fix the atomicity problem, we only need to add retry for system
reject.
+ public static boolean needRetryForSpecificCases(int statusCode, PlanNode
planNode) {
+ // To fix the atomicity problem, retry system rejection and a fenced
metadata lease that
+ // explicitly requires retry.
// 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();
+ if (statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()
+ && !planNode.isGeneratedByPipe()) {
+ return true;
+ }
+ return statusCode ==
TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()
+ && !planNode.isGeneratedByPipe();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
index a79eb652cf3..4b8ee72ffd4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
@@ -30,6 +30,7 @@ import
org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.exception.MetadataException;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.memory.IMemoryBlock;
import org.apache.iotdb.commons.memory.MemoryBlockType;
import org.apache.iotdb.commons.partition.DataPartition;
@@ -145,7 +146,8 @@ public class PartitionCache {
}
protected void failIfMetadataLeaseFenced() {
- MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced();
+ MetadataLeaseManager.getInstance()
+ .failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
}
// region database cache
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java
index 799f938edd4..29a66bdcdf0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternUtil;
@@ -74,7 +75,8 @@ public class TreeDeviceSchemaCacheManager {
}
void failIfMetadataLeaseFenced() {
- MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced();
+ MetadataLeaseManager.getInstance()
+ .failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
}
/** singleton pattern. */
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
index 9f00901da56..da198225243 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.schemaengine.lease;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
@@ -278,9 +279,10 @@ public class MetadataLeaseManager {
* refuse to serve it rather than risk validating writes/queries against
stale schema and
* producing dirty data.
*/
- public void failIfMetadataLeaseFenced() {
+ public void failIfMetadataLeaseFenced(final LeaseFencedRetryPolicy
leaseFencedRetryPolicy) {
if (isFenced()) {
- throw new
MetadataLeaseFencedException(DataNodeSchemaMessages.METADATA_LEASE_IS_FENCED);
+ throw new MetadataLeaseFencedException(
+ DataNodeSchemaMessages.METADATA_LEASE_IS_FENCED,
leaseFencedRetryPolicy);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
index 5b90104de83..c1bb53a8a90 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.exception.SemanticException;
import org.apache.iotdb.commons.schema.table.NonCommittableTsTable;
import org.apache.iotdb.commons.schema.table.PreDeleteTsTable;
@@ -102,8 +103,8 @@ public class DataNodeTableCache implements ITableCache {
return DataNodeTableCacheHolder.INSTANCE;
}
- void failIfMetadataLeaseFenced() {
- MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced();
+ void failIfMetadataLeaseFenced(final LeaseFencedRetryPolicy
leaseFencedRetryPolicy) {
+
MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
}
@Override
@@ -180,7 +181,7 @@ public class DataNodeTableCache implements ITableCache {
database = PathUtils.unQualifyDatabaseName(database);
readWriteLock.writeLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
specialStatusMap
.computeIfAbsent(database, k -> new ConcurrentHashMap<>())
.compute(
@@ -228,7 +229,7 @@ public class DataNodeTableCache implements ITableCache {
database = PathUtils.unQualifyDatabaseName(database);
readWriteLock.writeLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
// if rollback the drop table procedure, do nothing,
// wait for triggering the action of pull table from CN
final TsTable table = getTableFromSpecialStatusMap(database, tableName);
@@ -299,7 +300,7 @@ public class DataNodeTableCache implements ITableCache {
database = PathUtils.unQualifyDatabaseName(database);
readWriteLock.writeLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
final TsTable newTable = getTableFromSpecialStatusMap(database,
tableName);
if (Objects.isNull(newTable)) {
LOGGER.info(
@@ -422,7 +423,7 @@ public class DataNodeTableCache implements ITableCache {
public Map<String, Map<String, TsTable>> getTableSnapshot() {
readWriteLock.readLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
return databaseTableMap.entrySet().stream()
.collect(
Collectors.toMap(
@@ -444,7 +445,8 @@ public class DataNodeTableCache implements ITableCache {
@Override
public TsTable getTableInWrite(final String database, final String
tableName) {
- final TsTable result = getTableInCache(database, tableName);
+ final TsTable result =
+ getTableInCache(database, tableName,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
return Objects.nonNull(result) ? result : getTable(database, tableName,
false);
}
@@ -459,21 +461,30 @@ public class DataNodeTableCache implements ITableCache {
*/
@Override
public TsTable getTable(String database, final String tableName, final
boolean force) {
+ return getTable(database, tableName, force,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+ }
+
+ @Override
+ public TsTable getTable(
+ String database,
+ final String tableName,
+ final boolean force,
+ final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
database = PathUtils.unQualifyDatabaseName(database);
final AtomicReference<TableNodeStatus> tableStatusRef = new
AtomicReference<>();
final Map<String, Map<String, Long>> specialStatusMap =
- mayGetTableInSpecialStatusMap(database, tableName, tableStatusRef);
+ mayGetTableInSpecialStatusMap(database, tableName, tableStatusRef,
leaseFencedRetryPolicy);
if (Objects.nonNull(specialStatusMap) && !specialStatusMap.isEmpty()) {
Map<String, Map<String, TsTable>> fetchedTables =
getTablesInConfigNode(specialStatusMap, tableStatusRef.get());
if (tableStatusRef.get() == TableNodeStatus.USING) {
- updateUsingTable(fetchedTables, specialStatusMap);
+ updateUsingTable(fetchedTables, specialStatusMap,
leaseFencedRetryPolicy);
} else {
- updateDeleteTable(fetchedTables, database, tableName);
+ updateDeleteTable(fetchedTables, database, tableName,
leaseFencedRetryPolicy);
}
}
- final TsTable table = getTableInCache(database, tableName);
+ final TsTable table = getTableInCache(database, tableName,
leaseFencedRetryPolicy);
if (Objects.isNull(table) && force) {
CommonMetadataUtils.throwTableNotExistsException(database, tableName);
}
@@ -483,10 +494,11 @@ public class DataNodeTableCache implements ITableCache {
private Map<String, Map<String, Long>> mayGetTableInSpecialStatusMap(
final String database,
final String tableName,
- final AtomicReference<TableNodeStatus> tableNodeStatus) {
+ final AtomicReference<TableNodeStatus> tableNodeStatus,
+ final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
readWriteLock.readLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
final Map<String, Pair<TsTable, Long>> targetDatabaseMap =
specialStatusMap.get(database);
if (Objects.isNull(targetDatabaseMap)) {
return null;
@@ -560,10 +572,11 @@ public class DataNodeTableCache implements ITableCache {
private void updateUsingTable(
final Map<String, Map<String, TsTable>> fetchedTables,
- final Map<String, Map<String, Long>> previousVersions) {
+ final Map<String, Map<String, Long>> previousVersions,
+ final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
readWriteLock.writeLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
final AtomicBoolean isUpdated = new AtomicBoolean(false);
fetchedTables.forEach(
(qualifiedDatabase, tableInfoMap) -> {
@@ -618,10 +631,11 @@ public class DataNodeTableCache implements ITableCache {
private void updateDeleteTable(
Map<String, Map<String, TsTable>> fetchedTables,
String targetDatabase,
- final String targetTable) {
+ final String targetTable,
+ final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
readWriteLock.writeLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
boolean isUpdated = false;
boolean targetTableIsStillDeleting = false;
@@ -762,10 +776,13 @@ public class DataNodeTableCache implements ITableCache {
return modified ? builder.toString() :
DataNodeSchemaMessages.COMPARE_TABLE_NOT_MODIFIED;
}
- private TsTable getTableInCache(final String database, final String
tableName) {
+ private TsTable getTableInCache(
+ final String database,
+ final String tableName,
+ final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
readWriteLock.readLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
final TsTable result =
databaseTableMap.containsKey(database)
? databaseTableMap.get(database).get(tableName)
@@ -779,7 +796,7 @@ public class DataNodeTableCache implements ITableCache {
}
public boolean isDatabaseExist(final String database) {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
if (databaseTableMap.containsKey(database)) {
return true;
}
@@ -788,7 +805,7 @@ public class DataNodeTableCache implements ITableCache {
.containsKey(database)) {
readWriteLock.readLock().lock();
try {
- failIfMetadataLeaseFenced();
+ failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
databaseTableMap.computeIfAbsent(database, k -> new
ConcurrentHashMap<>());
return true;
} finally {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
index 26f67fb1129..13b05058955 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.schemaengine.table;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.schema.table.TsTable;
import javax.annotation.Nonnull;
@@ -53,6 +54,12 @@ public interface ITableCache {
TsTable getTable(String database, final String tableName, final boolean
force);
+ TsTable getTable(
+ String database,
+ final String tableName,
+ final boolean force,
+ final LeaseFencedRetryPolicy leaseFencedRetryPolicy);
+
Map<String, Map<String, TsTable>> getTableSnapshot();
String tryGetInternColumnName(
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 d2619a9de65..0d14d5b770d 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
@@ -31,6 +31,7 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.commons.exception.MetadataException;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.path.IFullPath;
import org.apache.iotdb.commons.path.MeasurementPath;
@@ -1740,7 +1741,9 @@ public class DataRegion implements IDataRegionForQuery {
t -> {
final String database = getDatabaseName();
- TsTable tsTable =
DataNodeTableCache.getInstance().getTable(database, t, false);
+ TsTable tsTable =
+ DataNodeTableCache.getInstance()
+ .getTable(database, t, false,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
if (tsTable == null) {
// There is a high probability that the leader node has been
executed and is currently
// located in the follower node.
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 c6ab83d7b6f..a1373c94603 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
@@ -150,6 +150,149 @@ public class DataRegionStateMachineTest {
Assert.assertEquals(2, planNode.getAcceptCount());
}
+ @Test
+ public void testMetadataLeaseFencedNoRetry() {
+ final DataRegionStateMachine stateMachine = new
DataRegionStateMachine(null);
+ final FixedStatusPlanNode planNode =
+ new
FixedStatusPlanNode(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode());
+
+ final TSStatus status = stateMachine.write(planNode);
+
+ Assert.assertEquals(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode(),
status.getCode());
+ Assert.assertEquals(1, planNode.getAcceptCount());
+ }
+
+ @Test
+ public void testMetadataLeaseFencedRetryRequiredRetriesAndFails() {
+ final DataRegionStateMachine stateMachine = new
DataRegionStateMachine(null);
+ final FixedStatusPlanNode planNode =
+ new
FixedStatusPlanNode(TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode());
+
+ final TSStatus status = stateMachine.write(planNode);
+
+ Assert.assertEquals(
+ TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode(),
status.getCode());
+ Assert.assertEquals(5, planNode.getAcceptCount());
+ }
+
+ @Test
+ public void testMetadataLeaseFencedRetryRequiredRetriesAndSucceeds() {
+ final DataRegionStateMachine stateMachine = new
DataRegionStateMachine(null);
+ final SequenceStatusPlanNode planNode =
+ new SequenceStatusPlanNode(
+ TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode(),
+ TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode());
+
+ final TSStatus status = stateMachine.write(planNode);
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(3, planNode.getAcceptCount());
+ }
+
+ private static class FixedStatusPlanNode extends PlanNode {
+
+ private final int statusCode;
+ private int acceptCount = 0;
+
+ private FixedStatusPlanNode(final int statusCode) {
+ super(new PlanNodeId("test"));
+ this.statusCode = statusCode;
+ }
+
+ private int getAcceptCount() {
+ return acceptCount;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R, C> R accept(final IPlanVisitor<R, C> visitor, final C context) {
+ ++acceptCount;
+ return (R) new TSStatus(statusCode);
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void addChild(final PlanNode child) {}
+
+ @Override
+ public PlanNode clone() {
+ return new FixedStatusPlanNode(statusCode);
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return NO_CHILD_ALLOWED;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected void serializeAttributes(final ByteBuffer byteBuffer) {}
+
+ @Override
+ protected void serializeAttributes(final DataOutputStream stream) throws
IOException {}
+ }
+
+ private static class SequenceStatusPlanNode extends PlanNode {
+
+ private final int[] statusCodes;
+ private int acceptCount = 0;
+
+ private SequenceStatusPlanNode(final int... statusCodes) {
+ super(new PlanNodeId("test"));
+ this.statusCodes = statusCodes;
+ }
+
+ private int getAcceptCount() {
+ return acceptCount;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R, C> R accept(final IPlanVisitor<R, C> visitor, final C context) {
+ final int code = statusCodes[Math.min(acceptCount, statusCodes.length -
1)];
+ ++acceptCount;
+ return (R) new TSStatus(code);
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void addChild(final PlanNode child) {}
+
+ @Override
+ public PlanNode clone() {
+ return new SequenceStatusPlanNode(statusCodes);
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return NO_CHILD_ALLOWED;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected void serializeAttributes(final ByteBuffer byteBuffer) {}
+
+ @Override
+ protected void serializeAttributes(final DataOutputStream stream) throws
IOException {}
+ }
+
private static class RetryControlledPlanNode extends PlanNode {
private final boolean markAsPipeOnFirstAccept;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
index e59873bf36a..7507d604838 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.schemaengine.lease;
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import com.google.common.util.concurrent.MoreExecutors;
@@ -47,10 +48,16 @@ public final class MetadataLeaseTestUtils {
}
public static void failIfMetadataLeaseFenced(final MetadataLeaseManager
manager) {
+ failIfMetadataLeaseFenced(manager, LeaseFencedRetryPolicy.NONE);
+ }
+
+ public static void failIfMetadataLeaseFenced(
+ final MetadataLeaseManager manager, final LeaseFencedRetryPolicy
leaseFencedRetryPolicy) {
manager.checkLeaseStatus();
if (manager.isFenced()) {
throw new MetadataLeaseFencedException(
- "Metadata lease is fenced. The local metadata cache is
unavailable.");
+ "Metadata lease is fenced. The local metadata cache is unavailable.",
+ leaseFencedRetryPolicy);
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java
index 5325f6e349c..00c5388e747 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.schemaengine.table;
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseTestUtils;
@@ -51,11 +52,12 @@ public class DataNodeTableCacheLeaseTest {
tableCache.invalidateAll();
Mockito.doAnswer(
invocation -> {
- MetadataLeaseTestUtils.failIfMetadataLeaseFenced(leaseManager);
+ MetadataLeaseTestUtils.failIfMetadataLeaseFenced(
+ leaseManager, invocation.getArgument(0));
return null;
})
.when(tableCache)
- .failIfMetadataLeaseFenced();
+ .failIfMetadataLeaseFenced(Mockito.any());
}
@After
@@ -82,9 +84,25 @@ public class DataNodeTableCacheLeaseTest {
assertLeaseFenced(() -> tableCache.commitUpdateTable("root.db", "t",
null));
}
+ @Test
+ public void stateMachineReadCarriesRetryPolicy() {
+ nowNanos.addAndGet((T_FENCE_MS + 1) * 1_000_000L);
+
+ final MetadataLeaseFencedException e =
+ assertThrows(
+ MetadataLeaseFencedException.class,
+ () ->
+ tableCache.getTable(
+ "root.db", "t", false,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS));
+
+ assertEquals(
+ TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode(),
e.getErrorCode());
+ }
+
private static void assertLeaseFenced(final Runnable runnable) {
final MetadataLeaseFencedException e =
assertThrows(MetadataLeaseFencedException.class, runnable::run);
- assertEquals(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode(),
e.getErrorCode());
+ assertEquals(
+ TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode(),
e.getErrorCode());
}
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
index b87f76bb18f..cc393f2f468 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
@@ -23,11 +23,24 @@ import org.apache.iotdb.rpc.TSStatusCode;
public class MetadataLeaseFencedException extends IoTDBRuntimeException {
- public MetadataLeaseFencedException(String message) {
- super(message, TSStatusCode.METADATA_LEASE_FENCED.getStatusCode());
+ public enum LeaseFencedRetryPolicy {
+ NONE,
+ RETRY_UNTIL_SUCCESS
}
- public MetadataLeaseFencedException(Throwable cause) {
- super(cause, TSStatusCode.METADATA_LEASE_FENCED.getStatusCode());
+ public MetadataLeaseFencedException(
+ String message, LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
+ super(message, getStatusCode(leaseFencedRetryPolicy));
+ }
+
+ public MetadataLeaseFencedException(
+ Throwable cause, LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
+ super(cause, getStatusCode(leaseFencedRetryPolicy));
+ }
+
+ private static int getStatusCode(LeaseFencedRetryPolicy
leaseFencedRetryPolicy) {
+ return leaseFencedRetryPolicy == LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS
+ ? TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()
+ : TSStatusCode.METADATA_LEASE_FENCED.getStatusCode();
}
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RetryUtils.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RetryUtils.java
index 56e9d7f401f..b07a95045c7 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RetryUtils.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RetryUtils.java
@@ -40,6 +40,7 @@ public class RetryUtils {
return statusCode == TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()
|| statusCode == TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()
|| statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()
+ || statusCode ==
TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()
|| statusCode == TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode();
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/StatusUtils.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/StatusUtils.java
index 246cf6d0ea1..f9cdb35b118 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/StatusUtils.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/StatusUtils.java
@@ -68,6 +68,7 @@ public class StatusUtils {
NEED_RETRY.add(TSStatusCode.TOO_MANY_CONCURRENT_QUERIES_ERROR.getStatusCode());
NEED_RETRY.add(TSStatusCode.SYNC_CONNECTION_ERROR.getStatusCode());
NEED_RETRY.add(TSStatusCode.PLAN_FAILED_NETWORK_PARTITION.getStatusCode());
+
NEED_RETRY.add(TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode());
}
/**
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/RetryStatusUtilsTest.java
similarity index 50%
copy from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
copy to
iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/RetryStatusUtilsTest.java
index b87f76bb18f..6ca93ed1bac 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/MetadataLeaseFencedException.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/RetryStatusUtilsTest.java
@@ -17,17 +17,32 @@
* under the License.
*/
-package org.apache.iotdb.commons.exception;
+package org.apache.iotdb.commons.utils;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.rpc.TSStatusCode;
-public class MetadataLeaseFencedException extends IoTDBRuntimeException {
+import org.junit.Assert;
+import org.junit.Test;
- public MetadataLeaseFencedException(String message) {
- super(message, TSStatusCode.METADATA_LEASE_FENCED.getStatusCode());
+public class RetryStatusUtilsTest {
+
+ @Test
+ public void testNeedRetryForMetadataLeaseFencedRetryRequired() {
+ Assert.assertTrue(
+ StatusUtils.needRetryHelper(
+ new
TSStatus(TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode())));
+ Assert.assertTrue(
+ RetryUtils.needRetryForWrite(
+
TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()));
}
- public MetadataLeaseFencedException(Throwable cause) {
- super(cause, TSStatusCode.METADATA_LEASE_FENCED.getStatusCode());
+ @Test
+ public void testNeedRetryForMetadataLeaseFenced() {
+ Assert.assertFalse(
+ StatusUtils.needRetryHelper(
+ new TSStatus(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode())));
+ Assert.assertFalse(
+
RetryUtils.needRetryForWrite(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode()));
}
}
