Subscription: Fixed multiple problems (#17418)

Author: Caideyipi

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

@@ -220,26 +220,21 @@ public void validatePipePluginUsageByTopic(String pipePluginName) throws Subscri
     }
   }
 
-  public void validatePipePluginUsageByTopicInternal(String pipePluginName)
+  private void validatePipePluginUsageByTopicInternal(String pipePluginName)
       throws SubscriptionException {
-    acquireReadLock();
-    try {
-      topicMetaKeeper
-          .getAllTopicMeta()
-          .forEach(
-              meta -> {
-                if (pipePluginName.equals(meta.getConfig().getAttribute().get("processor"))) {
-                  final String exceptionMessage =
-                      String.format(
-                          "PipePlugin '%s' is already used by Topic '%s' as a processor.",
-                          pipePluginName, meta.getTopicName());
-                  LOGGER.warn(exceptionMessage);
-                  throw new SubscriptionException(exceptionMessage);
-                }
-              });
-    } finally {
-      releaseReadLock();
-    }
+    topicMetaKeeper
+        .getAllTopicMeta()
+        .forEach(
+            meta -> {
+              if (pipePluginName.equals(meta.getConfig().getAttribute().get("processor"))) {
+                final String exceptionMessage =
+                    String.format(
+                        "PipePlugin '%s' is already used by Topic '%s' as a processor.",
+                        pipePluginName, meta.getTopicName());
+                LOGGER.warn(exceptionMessage);
+                throw new SubscriptionException(exceptionMessage);
+              }
+            });
   }
 
   public void validateBeforeAlteringTopic(TopicMeta topicMeta) throws SubscriptionException {
@@ -335,21 +330,25 @@ public TSStatus createTopic(CreateTopicPlan plan) {
   public TSStatus alterTopic(AlterTopicPlan plan) {
     acquireWriteLock();
     try {
-      topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName());
-      topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(), plan.getTopicMeta());
-      return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+      return alterTopicInternal(plan);
     } finally {
       releaseWriteLock();
     }
   }
 
-  public TSStatus alterMultipleTopics(AlterMultipleTopicsPlan plan) {
+  private TSStatus alterTopicInternal(final AlterTopicPlan plan) {
+    topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName());
+    topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(), plan.getTopicMeta());
+    return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+  }
+
+  public TSStatus alterMultipleTopics(final AlterMultipleTopicsPlan plan) {
     acquireWriteLock();
     try {
-      TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+      final TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
       status.setSubStatus(new ArrayList<>());
-      for (AlterTopicPlan subPlan : plan.getSubPlans()) {
-        TSStatus innerStatus = alterTopic(subPlan);
+      for (final AlterTopicPlan subPlan : plan.getSubPlans()) {
+        final TSStatus innerStatus = alterTopicInternal(subPlan);
         status.getSubStatus().add(innerStatus);
         if (innerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
           status.setCode(TSStatusCode.ALTER_TOPIC_ERROR.getStatusCode());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java

@@ -70,10 +70,9 @@ public DeletionResource(
   }
 
   public synchronized void decreaseReference() {
-    if (pipeTaskReferenceCount.get() == 1) {
+    if (pipeTaskReferenceCount.decrementAndGet() == 0) {
       removeSelf();
     }
-    pipeTaskReferenceCount.decrementAndGet();
   }
 
   public void removeSelf() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java

@@ -279,7 +279,7 @@ private int getPrefetchingQueueCountInternal() {
    */
   private static class Cache<T> {
 
-    private T value;
+    private volatile T value;
     private volatile boolean valid = false;
     private final Supplier<T> supplier;
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java

@@ -96,7 +96,6 @@ private void handleSingleConsumerGroupMetaChangesInternal(
 
     // if consumer group meta does not exist on local agent
     if (Objects.isNull(metaInAgent)) {
-      consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId);
       consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId, metaFromCoordinator);
       SubscriptionAgent.broker().createBrokerIfNotExist(consumerGroupId);
       return;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java

@@ -149,10 +149,12 @@ public boolean isTopicExisted(final String topicName) {
   public String getTopicFormat(final String topicName) {
     acquireReadLock();
     try {
-      return topicMetaKeeper
-          .getTopicMeta(topicName)
-          .getConfig()
-          .getStringOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE);
+      return topicMetaKeeper.containsTopicMeta(topicName)
+          ? topicMetaKeeper
+              .getTopicMeta(topicName)
+              .getConfig()
+              .getStringOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE)
+          : null;
     } finally {
       releaseReadLock();
     }
@@ -161,10 +163,12 @@ public String getTopicFormat(final String topicName) {
   public String getTopicMode(final String topicName) {
     acquireReadLock();
     try {
-      return topicMetaKeeper
-          .getTopicMeta(topicName)
-          .getConfig()
-          .getStringOrDefault(TopicConstant.MODE_KEY, TopicConstant.MODE_DEFAULT_VALUE);
+      return topicMetaKeeper.containsTopicMeta(topicName)
+          ? topicMetaKeeper
+              .getTopicMeta(topicName)
+              .getConfig()
+              .getStringOrDefault(TopicConstant.MODE_KEY, TopicConstant.MODE_DEFAULT_VALUE)
+          : null;
     } finally {
       releaseReadLock();
     }
@@ -174,6 +178,7 @@ public Map<String, TopicConfig> getTopicConfigs(final Set<String> topicNames) {
     acquireReadLock();
     try {
       return topicNames.stream()
+          .filter(topicMetaKeeper::containsTopicMeta)
           .collect(
               Collectors.toMap(
                   topicName -> topicName,

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java

@@ -34,6 +34,7 @@
 import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -146,11 +147,13 @@ public void addConsumer(final ConsumerMeta consumerMeta) {
 
   public void removeConsumer(final String consumerId) {
     consumerIdToConsumerMeta.remove(consumerId);
-    for (final Map.Entry<String, Set<String>> entry :
-        topicNameToSubscribedConsumerIdSet.entrySet()) {
+    final Iterator<Map.Entry<String, Set<String>>> iterator =
+        topicNameToSubscribedConsumerIdSet.entrySet().iterator();
+    while (iterator.hasNext()) {
+      final Map.Entry<String, Set<String>> entry = iterator.next();
       entry.getValue().remove(consumerId);
       if (entry.getValue().isEmpty()) {
-        topicNameToSubscribedConsumerIdSet.remove(entry.getKey());
+        iterator.remove();
       }
     }
   }