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 384f38028e84a..1cb9c96771e24 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 by timecho DB + 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 d7900cf2388b4..6c0366c546fe8 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.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 CompletableFuture applyTransaction(TransactionContext trx) { 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 CompletableFuture applyTransaction(TransactionContext trx) { 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 CompletableFuture applyTransaction(TransactionContext trx) { 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 bd48ed25ae759..96bbb6ccc16ba 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 @@ 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 TSStatus visitInsertRow(InsertRowNode node, DataRegion 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 TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion } } return firstStatus; + } catch (final MetadataLeaseFencedException e) { + return RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); } } @@ -170,6 +175,8 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion 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 TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion } } return firstStatus; + } catch (MetadataLeaseFencedException e) { + return RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); } } @@ -243,6 +252,8 @@ public TSStatus visitInsertRowsOfOneDevice( } } 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 8887054e2f29c..051dcbb277197 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 @@ protected TSStatus write(PlanNode planNode) { 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 File getSnapshotRoot() { } } - 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 a79eb652cf324..4b8ee72ffd484 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.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 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 799f938edd4fd..29a66bdcdf0ad 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 static TreeDeviceSchemaCacheManager getInstance() { } 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 9f00901da5688..da198225243da 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 @@ 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 @@ void checkLeaseStatus() { * 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 5b90104de838f..c1bb53a8a909e 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.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 static ITableCache getInstance() { return DataNodeTableCacheHolder.INSTANCE; } - void failIfMetadataLeaseFenced() { - MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced(); + void failIfMetadataLeaseFenced(final LeaseFencedRetryPolicy leaseFencedRetryPolicy) { + MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced(leaseFencedRetryPolicy); } @Override @@ -180,7 +181,7 @@ public void preUpdateTable(String database, final TsTable table, final String ol 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 void rollbackUpdateTable(String database, final String tableName, final S 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 void commitUpdateTable( 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 long getInstanceVersion() { public Map> getTableSnapshot() { readWriteLock.readLock().lock(); try { - failIfMetadataLeaseFenced(); + failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); return databaseTableMap.entrySet().stream() .collect( Collectors.toMap( @@ -444,7 +445,8 @@ public Map> getTableSnapshot() { @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 TsTable getTable(final String database, final String tableName) { */ @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 tableStatusRef = new AtomicReference<>(); final Map> specialStatusMap = - mayGetTableInSpecialStatusMap(database, tableName, tableStatusRef); + mayGetTableInSpecialStatusMap(database, tableName, tableStatusRef, leaseFencedRetryPolicy); if (Objects.nonNull(specialStatusMap) && !specialStatusMap.isEmpty()) { Map> 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 TsTable getTable(String database, final String tableName, final boolean f private Map> mayGetTableInSpecialStatusMap( final String database, final String tableName, - final AtomicReference tableNodeStatus) { + final AtomicReference tableNodeStatus, + final LeaseFencedRetryPolicy leaseFencedRetryPolicy) { readWriteLock.readLock().lock(); try { - failIfMetadataLeaseFenced(); + failIfMetadataLeaseFenced(leaseFencedRetryPolicy); final Map> targetDatabaseMap = specialStatusMap.get(database); if (Objects.isNull(targetDatabaseMap)) { return null; @@ -560,10 +572,11 @@ private Map> getTablesInConfigNode( private void updateUsingTable( final Map> fetchedTables, - final Map> previousVersions) { + final Map> 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 @@ private void updateUsingTable( private void updateDeleteTable( Map> 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 @@ private String compareTable(final TsTable oldTable, final TsTable newTable) { 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 @@ private TsTable getTableInCache(final String database, final String tableName) { } public boolean isDatabaseExist(final String database) { - failIfMetadataLeaseFenced(); + failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS); if (databaseTableMap.containsKey(database)) { return true; } @@ -788,7 +805,7 @@ public boolean isDatabaseExist(final String database) { .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 26f67fb112956..13b0505895519 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 @@ void commitUpdateTable( TsTable getTable(String database, final String tableName, final boolean force); + TsTable getTable( + String database, + final String tableName, + final boolean force, + final LeaseFencedRetryPolicy leaseFencedRetryPolicy); + Map> 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 d2619a9de65bf..0d14d5b770d7b 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.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 @@ private void registerToTsFile(InsertNode node, TsFileProcessor tsFileProcessor) 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 c6ab83d7b6f97..a1373c9460343 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 void testNonPipeWriteProcessRejectCanStillRetry() { 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 accept(final IPlanVisitor visitor, final C context) { + ++acceptCount; + return (R) new TSStatus(statusCode); + } + + @Override + public List 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 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 accept(final IPlanVisitor visitor, final C context) { + final int code = statusCodes[Math.min(acceptCount, statusCodes.length - 1)]; + ++acceptCount; + return (R) new TSStatus(code); + } + + @Override + public List 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 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 e59873bf36a87..7507d604838a1 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 static boolean isFenced(final MetadataLeaseManager manager) { } 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 5325f6e349ce5..8604faf6d4867 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 void setUp() { tableCache.invalidateAll(); Mockito.doAnswer( invocation -> { - MetadataLeaseTestUtils.failIfMetadataLeaseFenced(leaseManager); + MetadataLeaseTestUtils.failIfMetadataLeaseFenced( + leaseManager, invocation.getArgument(0)); return null; }) .when(tableCache) - .failIfMetadataLeaseFenced(); + .failIfMetadataLeaseFenced(Mockito.any()); } @After @@ -82,6 +84,21 @@ public void fencedLeaseFailsClosedForUpdateApis() { 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); 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 b87f76bb18f12..cc393f2f46858 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 @@ 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 56e9d7f401f2a..b07a95045c732 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 static boolean needRetryForWrite(int statusCode) { 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 246cf6d0ea1d0..f9cdb35b11880 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 @@ private 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/test/java/org/apache/iotdb/commons/utils/RetryStatusUtilsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/RetryStatusUtilsTest.java new file mode 100644 index 0000000000000..6ca93ed1bac9a --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/RetryStatusUtilsTest.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.utils; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.Test; + +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())); + } + + @Test + public void testNeedRetryForMetadataLeaseFenced() { + Assert.assertFalse( + StatusUtils.needRetryHelper( + new TSStatus(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode()))); + Assert.assertFalse( + RetryUtils.needRetryForWrite(TSStatusCode.METADATA_LEASE_FENCED.getStatusCode())); + } +}