refactor thread model

Author: DanielWang2035

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java

@@ -33,7 +33,7 @@
 public class PipeSubtaskExecutorManager {
   private final PipeProcessorSubtaskExecutor processorExecutor;
   private final Supplier<PipeSinkSubtaskExecutor> connectorExecutorSupplier;
-  private final SubscriptionSubtaskExecutor subscriptionExecutor;
+  private volatile SubscriptionSubtaskExecutor subscriptionExecutor;
 
   public PipeProcessorSubtaskExecutor getProcessorExecutor() {
     return processorExecutor;
@@ -49,6 +49,7 @@ public IoTConsensusV2SubtaskExecutor getConsensusExecutor() {
   }
 
   public SubscriptionSubtaskExecutor getSubscriptionExecutor() {
+    ensureSubscriptionExecutors();
     return subscriptionExecutor;
   }
 
@@ -57,15 +58,28 @@ public SubscriptionSubtaskExecutor getSubscriptionExecutor() {
   private PipeSubtaskExecutorManager() {
     processorExecutor = new PipeProcessorSubtaskExecutor();
     connectorExecutorSupplier = PipeSinkSubtaskExecutor::new;
-    subscriptionExecutor =
-        SubscriptionConfig.getInstance().getSubscriptionEnabled()
-            ? new SubscriptionSubtaskExecutor()
-            : null;
+    ensureSubscriptionExecutors();
     // IoTV2 uses global singleton executor pool.
     IoTV2GlobalComponentContainer.getInstance()
         .setConsensusExecutor(new IoTConsensusV2SubtaskExecutor());
   }
 
+  public synchronized void ensureSubscriptionExecutors() {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return;
+    }
+    if (subscriptionExecutor == null || subscriptionExecutor.isShutdown()) {
+      subscriptionExecutor = new SubscriptionSubtaskExecutor();
+    }
+  }
+
+  public synchronized void shutdownSubscriptionExecutors() {
+    if (subscriptionExecutor != null) {
+      subscriptionExecutor.shutdown();
+      subscriptionExecutor = null;
+    }
+  }
+
   private static class PipeTaskExecutorHolder {
     private static PipeSubtaskExecutorManager instance = null;
   }

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

@@ -29,6 +29,7 @@
 import org.apache.iotdb.db.subscription.broker.consensus.ConsensusSubscriptionSetupHandler;
 import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
 import org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager;
+import org.apache.iotdb.db.subscription.task.execution.ConsensusSubscriptionPrefetchExecutorManager;
 import org.apache.iotdb.db.subscription.task.subtask.SubscriptionSinkSubtask;
 import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
@@ -229,6 +230,7 @@ public void seek(
     final ConsensusSubscriptionBroker consensusBroker =
         consumerGroupIdToConsensusBroker.get(consumerGroupId);
     if (Objects.nonNull(consensusBroker) && consensusBroker.hasQueue(topicName)) {
+      ensureConsensusSeekRuntimeAvailable(consumerGroupId, topicName, "seek");
       if (seekType != PipeSubscribeSeekReq.SEEK_TO_BEGINNING
           && seekType != PipeSubscribeSeekReq.SEEK_TO_END) {
         final String errorMessage =
@@ -261,6 +263,7 @@ public void seekToTopicProgress(
     final ConsensusSubscriptionBroker consensusBroker =
         consumerGroupIdToConsensusBroker.get(consumerGroupId);
     if (Objects.nonNull(consensusBroker) && consensusBroker.hasQueue(topicName)) {
+      ensureConsensusSeekRuntimeAvailable(consumerGroupId, topicName, "seek(topicProgress)");
       consensusBroker.seek(topicName, topicProgress);
       return;
     }
@@ -283,6 +286,7 @@ public void seekAfterTopicProgress(
     final ConsensusSubscriptionBroker consensusBroker =
         consumerGroupIdToConsensusBroker.get(consumerGroupId);
     if (Objects.nonNull(consensusBroker) && consensusBroker.hasQueue(topicName)) {
+      ensureConsensusSeekRuntimeAvailable(consumerGroupId, topicName, "seekAfter(topicProgress)");
       consensusBroker.seekAfter(topicName, topicProgress);
       return;
     }
@@ -296,6 +300,20 @@ public void seekAfterTopicProgress(
     throw new SubscriptionException(errorMessage);
   }
 
+  private void ensureConsensusSeekRuntimeAvailable(
+      final String consumerGroupId, final String topicName, final String operation) {
+    if (!ConsensusSubscriptionPrefetchExecutorManager.getInstance().isStarted()
+        || SubscriptionAgent.runtime().isShutdown()) {
+      final String errorMessage =
+          String.format(
+              "Subscription: consensus %s is unavailable because subscription runtime is stopped, "
+                  + "consumerGroup=%s, topic=%s",
+              operation, consumerGroupId, topicName);
+      LOGGER.warn(errorMessage);
+      throw new SubscriptionException(errorMessage);
+    }
+  }
+
   public boolean isCommitContextOutdated(final SubscriptionCommitContext commitContext) {
     final String consumerGroupId = commitContext.getConsumerGroupId();
     final String topicName = commitContext.getTopicName();
@@ -533,6 +551,12 @@ public void applyRuntimeStateForRegion(
     }
   }
 
+  public void abortConsensusPendingSeeksForRuntimeStop() {
+    for (final ConsensusSubscriptionBroker broker : consumerGroupIdToConsensusBroker.values()) {
+      broker.abortPendingSeeksForRuntimeStop();
+    }
+  }
+
   public void updateCompletedTopicNames(final String consumerGroupId, final String topicName) {
     final SubscriptionBroker pipeBroker = consumerGroupIdToPipeBroker.get(consumerGroupId);
     if (Objects.isNull(pipeBroker)) {
@@ -584,12 +608,10 @@ public void removePrefetchingQueue(final String consumerGroupId, final String to
   }
 
   public boolean executePrefetch(final String consumerGroupId, final String topicName) {
-    // Try consensus broker first
-    final ConsensusSubscriptionBroker consensusBroker =
-        consumerGroupIdToConsensusBroker.get(consumerGroupId);
-    if (Objects.nonNull(consensusBroker) && consensusBroker.hasQueue(topicName)) {
-      return consensusBroker.executePrefetch(topicName);
+    if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName)) {
+      return false;
     }
+
     // Fall back to pipe broker
     final SubscriptionBroker pipeBroker = consumerGroupIdToPipeBroker.get(consumerGroupId);
     if (Objects.isNull(pipeBroker)) {

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

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.service.IService;
 import org.apache.iotdb.commons.service.ServiceType;
 import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import org.apache.iotdb.db.subscription.task.execution.ConsensusSubscriptionPrefetchExecutorManager;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -67,6 +68,7 @@ public void start() throws StartupException {
     }
 
     SubscriptionConfig.getInstance().printAllConfigs();
+    ConsensusSubscriptionPrefetchExecutorManager.getInstance().start();
 
     SubscriptionAgentLauncher.launchSubscriptionTopicAgent();
     SubscriptionAgentLauncher.launchSubscriptionConsumerAgent();
@@ -80,8 +82,9 @@ public void stop() {
       return;
     }
     isShutdown.set(true);
-
-    // let PipeDataNodeRuntimeAgent to drop all related pipe tasks
+    SubscriptionAgent.broker().abortConsensusPendingSeeksForRuntimeStop();
+    ConsensusSubscriptionPrefetchExecutorManager.getInstance().stop();
+    SubscriptionAgent.broker().abortConsensusPendingSeeksForRuntimeStop();
   }
 
   public boolean isShutdown() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java

@@ -380,18 +380,10 @@ private void seekQueueToRegionProgress(
 
   @Override
   public boolean executePrefetch(final String topicName) {
-    final List<ConsensusPrefetchingQueue> queues =
-        topicNameToConsensusPrefetchingQueues.get(topicName);
-    if (Objects.isNull(queues) || queues.isEmpty()) {
-      return false;
-    }
-    boolean anyPrefetched = false;
-    for (final ConsensusPrefetchingQueue q : queues) {
-      if (!q.isClosed() && q.executePrefetch()) {
-        anyPrefetched = true;
-      }
-    }
-    return anyPrefetched;
+    // Consensus prefetch is fully driven by queue-local wakeup sources and the dedicated delayed
+    // scheduler. This interface remains only to satisfy the shared broker contract used by
+    // pipe-based subscription.
+    return false;
   }
 
   @Override
@@ -721,6 +713,17 @@ public void applyRuntimeStateForRegion(
     }
   }
 
+  public void abortPendingSeeksForRuntimeStop() {
+    for (final List<ConsensusPrefetchingQueue> queues :
+        topicNameToConsensusPrefetchingQueues.values()) {
+      for (final ConsensusPrefetchingQueue q : queues) {
+        if (!q.isClosed()) {
+          q.abortPendingSeekForRuntimeStop();
+        }
+      }
+    }
+  }
+
   @Override
   public void removeQueue(final String topicName) {
     final List<ConsensusPrefetchingQueue> queues =