This is an automated email from the ASF dual-hosted git repository. JackieTien97 pushed a commit to branch rc/2.0.11 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ecec8eba0aff044d5c1c833b6fbe60393b309a0c 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 7f29ab552df..699b392a80e 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; @@ -1686,7 +1687,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())); } }
