support the retry for the execution the insert path in statemachine with DN metadata fenced and retry for coordinator

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),

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<Message> 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<Message> 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<Message> 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;
     }
   }
 

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());
     }
   }
 

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();
   }
 }

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

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. */

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);
     }
   }
 

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<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 Map<String, Map<String, TsTable>> 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<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 TsTable getTable(String database, final String tableName, final boolean f
   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 @@ private Map<String, Map<String, TsTable>> getTablesInConfigNode(
 
   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 @@ private void updateUsingTable(
   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 @@ 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 {