Fix write retry, pipe fallback, and snapshot recovery edge cases

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -47,6 +47,7 @@
 import org.apache.iotdb.commons.conf.ConfigurationFileUtils;
 import org.apache.iotdb.commons.conf.IoTDBConstant;
 import org.apache.iotdb.commons.conf.TrimProperties;
+import org.apache.iotdb.commons.consensus.DataRegionId;
 import org.apache.iotdb.commons.enums.RepairDataPartitionTableProgressState;
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.exception.MetadataException;
@@ -66,6 +67,7 @@
 import org.apache.iotdb.commons.schema.tree.AlterTimeSeriesOperationType;
 import org.apache.iotdb.commons.schema.ttl.TTLCache;
 import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper;
 import org.apache.iotdb.commons.utils.AuthUtils;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.commons.utils.StatusUtils;
@@ -2679,19 +2681,60 @@ public TGetCommitProgressResp getCommitProgress(TGetCommitProgressReq req) {
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       return new TGetCommitProgressResp(status);
     }
-    final String keyPrefix =
-        req.getConsumerGroupId() + "##" + req.getTopicName() + "##" + req.getRegionId() + "##";
-    final org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper keeper =
+    final CommitProgressKeeper keeper =
         subscriptionManager
             .getSubscriptionCoordinator()
             .getSubscriptionInfo()
             .getCommitProgressKeeper();
-    final Map<WriterId, WriterProgress> mergedWriterPositions = new LinkedHashMap<>();
+    final RegionProgress mergedRegionProgress =
+        mergeCommitProgress(
+            keeper.getAllRegionProgress(),
+            req.getConsumerGroupId(),
+            req.getTopicName(),
+            req.getRegionId());
+    final TGetCommitProgressResp resp =
+        new TGetCommitProgressResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+    if (!mergedRegionProgress.getWriterPositions().isEmpty()) {
+      resp.setCommittedRegionProgress(serializeRegionProgress(mergedRegionProgress));
+    }
+    return resp;
+  }
 
-    for (final Map.Entry<String, ByteBuffer> entry : keeper.getAllRegionProgress().entrySet()) {
-      if (!entry.getKey().startsWith(keyPrefix)) {
+  static RegionProgress mergeCommitProgress(
+      final Map<String, ByteBuffer> allRegionProgress,
+      final String consumerGroupId,
+      final String topicName,
+      final int regionId) {
+    final String regionIdString = new DataRegionId(regionId).toString();
+    final Map<WriterId, WriterProgress> mergedWriterPositions = new LinkedHashMap<>();
+    final boolean hasVersionedProgress =
+        mergeCommitProgressForPrefix(
+            allRegionProgress,
+            CommitProgressKeeper.generateRegionKeyPrefix(
+                consumerGroupId, topicName, regionIdString),
+            mergedWriterPositions);
+    if (!hasVersionedProgress
+        || CommitProgressKeeper.isLegacyKeyUnambiguous(
+            consumerGroupId, topicName, regionIdString)) {
+      mergeCommitProgressForPrefix(
+          allRegionProgress,
+          CommitProgressKeeper.generateLegacyRegionKeyPrefix(
+              consumerGroupId, topicName, regionIdString),
+          mergedWriterPositions);
+    }
+    return new RegionProgress(mergedWriterPositions);
+  }
+
+  private static boolean mergeCommitProgressForPrefix(
+      final Map<String, ByteBuffer> allRegionProgress,
+      final String keyPrefix,
+      final Map<WriterId, WriterProgress> mergedWriterPositions) {
+    boolean hasMatchingProgress = false;
+    for (final Map.Entry<String, ByteBuffer> entry : allRegionProgress.entrySet()) {
+      if (!CommitProgressKeeper.isValidDataNodeProgressKey(entry.getKey(), keyPrefix)) {
         continue;
       }
+      hasMatchingProgress = true;
       final RegionProgress regionProgress = deserializeRegionProgress(entry.getValue());
       if (Objects.isNull(regionProgress)) {
         continue;
@@ -2705,13 +2748,7 @@ public TGetCommitProgressResp getCommitProgress(TGetCommitProgressReq req) {
                 compareWriterProgress(newProgress, oldProgress) > 0 ? newProgress : oldProgress);
       }
     }
-    final TGetCommitProgressResp resp =
-        new TGetCommitProgressResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
-    if (!mergedWriterPositions.isEmpty()) {
-      resp.setCommittedRegionProgress(
-          serializeRegionProgress(new RegionProgress(mergedWriterPositions)));
-    }
-    return resp;
+    return hasMatchingProgress;
   }
 
   private static RegionProgress deserializeRegionProgress(final ByteBuffer buffer) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java

@@ -285,7 +285,7 @@ public void processLoadSnapshot(File snapshotDir) throws IOException {
     File snapshotFile = new File(snapshotDir, SNAPSHOT_FILENAME);
 
     if (!snapshotFile.exists()) {
-      // do nothing if the snapshot file is not existed
+      clear();
       return;
     }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
 import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
 import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -127,6 +128,11 @@ public void decreaseListenerReference(final PipeParameters parameters)
     regionListener.decreaseReference(parameters);
   }
 
+  public void reconcileListenerReferences(final Iterable<PipeMeta> pipeMetas)
+      throws IllegalPathException {
+    regionListener.reconcileReferences(pipeMetas);
+  }
+
   /** Notify the region listener that the leader is ready to allow pipe operations. */
   public void notifyLeaderReady() {
     regionListener.notifyLeaderReady();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.confignode.manager.pipe.agent.runtime;
 
 import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
 import org.apache.iotdb.confignode.manager.pipe.source.ConfigRegionListeningFilter;
 import org.apache.iotdb.confignode.manager.pipe.source.ConfigRegionListeningQueue;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -57,6 +58,24 @@ public synchronized void decreaseReference(final PipeParameters parameters)
     }
   }
 
+  public synchronized void reconcileReferences(final Iterable<PipeMeta> pipeMetas)
+      throws IllegalPathException {
+    int referenceCount = 0;
+    for (final PipeMeta pipeMeta : pipeMetas) {
+      if (!ConfigRegionListeningFilter.parseListeningPlanTypeSet(
+              pipeMeta.getStaticMeta().getSourceParameters())
+          .isEmpty()) {
+        referenceCount++;
+      }
+    }
+    listeningQueueReferenceCount = referenceCount;
+    if (referenceCount == 0) {
+      listeningQueue.close();
+    } else {
+      listeningQueue.open();
+    }
+  }
+
   public synchronized boolean isLeaderReady() {
     return isLeaderReady.get();
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.common.rpc.thrift.TSchemaNode;
 import org.apache.iotdb.commons.auth.AuthException;
+import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.schema.node.MNodeType;
 import org.apache.iotdb.commons.schema.ttl.TTLCache;
@@ -804,10 +805,17 @@ public boolean loadSnapshot(final File latestSnapshotRootDir) {
               }
             });
     if (result.get()) {
-      pipeInfo.getPipeTaskInfo().enrichPipeMetasWithRootUserForCompatibility();
-      LOGGER.info(
-          ConfigNodeMessages.CONFIGNODESNAPSHOT_LOAD_SNAPSHOT_SUCCESS_LATESTSNAPSHOTROOTDIR,
-          latestSnapshotRootDir);
+      try {
+        PipeConfigNodeAgent.runtime()
+            .reconcileListenerReferences(pipeInfo.getPipeTaskInfo().getPipeMetaList());
+        pipeInfo.getPipeTaskInfo().enrichPipeMetasWithRootUserForCompatibility();
+        LOGGER.info(
+            ConfigNodeMessages.CONFIGNODESNAPSHOT_LOAD_SNAPSHOT_SUCCESS_LATESTSNAPSHOTROOTDIR,
+            latestSnapshotRootDir);
+      } catch (final IllegalPathException e) {
+        result.set(false);
+        LOGGER.error(ConfigNodeMessages.LOAD_SNAPSHOT_ERROR, e);
+      }
     }
     // Propagate any snapshot-load failure so callers (e.g. the AddPeer flow) do not treat a
     // partially or wholly failed load as success.

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java

@@ -311,11 +311,6 @@ public void processLoadSnapshot(final File snapshotDir) throws IOException {
 
     try {
       pipeTaskInfo.processLoadSnapshot(snapshotDir);
-
-      for (final PipeMeta pipeMeta : pipeTaskInfo.getPipeMetaList()) {
-        PipeConfigNodeAgent.runtime()
-            .increaseListenerReference(pipeMeta.getStaticMeta().getSourceParameters());
-      }
     } catch (final Exception ex) {
       LOGGER.error(ConfigNodeMessages.FAILED_TO_LOAD_PIPE_TASK_INFO_FROM_SNAPSHOT, ex);
       loadPipeTaskInfoException = ex;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java

@@ -259,6 +259,7 @@ public Map<String, TThrottleQuota> getThrottleQuotaLimit() {
 
   public void clear() {
     spaceQuotaLimit.clear();
+    spaceQuotaUsage.clear();
     throttleQuotaLimit.clear();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTable.java

@@ -149,6 +149,8 @@ public void processLoadSnapshot(File snapshotDir) throws IOException {
     try {
       File snapshotFile = new File(snapshotDir, SNAPSHOT_FILENAME);
       if (!snapshotFile.exists()) {
+        // Empty preset tables are represented by the absence of a snapshot file.
+        templatePreSetMap.clear();
         return;
       }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplateTable.java

@@ -262,16 +262,22 @@ public void processLoadSnapshot(File snapshotDir) throws IOException {
         BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream)) {
       // Load snapshot of template
       this.templateMap.clear();
+      this.templateIdMap.clear();
       deserialize(bufferedInputStream);
-      bufferedInputStream.close();
-      fileInputStream.close();
     } finally {
       templateReadWriteLock.writeLock().unlock();
     }
   }
 
   @TestOnly
   public void clear() {
-    this.templateMap.clear();
+    templateReadWriteLock.writeLock().lock();
+    try {
+      this.templateMap.clear();
+      this.templateIdMap.clear();
+      this.templateIdGenerator.set(0);
+    } finally {
+      templateReadWriteLock.writeLock().unlock();
+    }
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java

@@ -848,9 +848,19 @@ public boolean isTopicSubscribedByConsumerGroup(
   public TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) {
     acquireWriteLock();
     try {
-      ConsumerGroupMeta consumerGroupMeta = plan.getConsumerGroupMeta();
+      final ConsumerGroupMeta consumerGroupMeta = plan.getConsumerGroupMeta();
       if (Objects.nonNull(consumerGroupMeta)) {
-        String consumerGroupId = consumerGroupMeta.getConsumerGroupId();
+        final String consumerGroupId = consumerGroupMeta.getConsumerGroupId();
+        final ConsumerGroupMeta currentConsumerGroupMeta =
+            consumerGroupMetaKeeper.containsConsumerGroupMeta(consumerGroupId)
+                ? consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId)
+                : null;
+        if (Objects.nonNull(currentConsumerGroupMeta)) {
+          ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta, consumerGroupMeta)
+              .forEach(
+                  topicName ->
+                      commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName));
+        }
         consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId);
         if (!consumerGroupMeta.isEmpty()) {
           consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId, consumerGroupMeta);