[Subscription] Skip runtime when subscription is disabled

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java

@@ -3635,7 +3635,9 @@ public void handleClientExit() {
     }
     PipeDataNodeAgent.receiver().thrift().handleClientExit();
     PipeDataNodeAgent.receiver().legacy().handleClientExit();
-    SubscriptionAgent.receiver().handleClientExit();
+    if (COMMON_CONFIG.getSubscriptionEnabled()) {
+      SubscriptionAgent.receiver().handleClientExit();
+    }
   }
 
   /**

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

@@ -1627,6 +1627,11 @@ public TPushConsumerGroupMetaResp pushSingleConsumerGroupMeta(
 
   @Override
   public TPullCommitProgressResp pullCommitProgress(TPullCommitProgressReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TPullCommitProgressResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))
+          .setCommitRegionProgress(Collections.emptyMap());
+    }
+
     try {
       final int dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
       final Map<String, ByteBuffer> regionProgress =
@@ -1643,6 +1648,10 @@ public TPullCommitProgressResp pullCommitProgress(TPullCommitProgressReq req) {
 
   @Override
   public TSStatus syncSubscriptionProgress(TSyncSubscriptionProgressReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    }
+
     try {
       SubscriptionAgent.broker()
           .receiveSubscriptionProgress(
@@ -1664,6 +1673,10 @@ public TSStatus syncSubscriptionProgress(TSyncSubscriptionProgressReq req) {
 
   @Override
   public TSStatus pushSubscriptionRuntime(TPushSubscriptionRuntimeReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    }
+
     try {
       for (final TSubscriptionRuntimeStateEntry runtimeStateEntry : req.getRuntimeStates()) {
         ConsensusSubscriptionSetupHandler.applyRuntimeState(
@@ -2437,8 +2450,12 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th
 
   @Override
   public TSStatus updateRegionCache(TRegionRouteReq req) {
-    boolean result = ClusterPartitionFetcher.getInstance().updateRegionCache(req);
-    if (result) {
+    final boolean result = ClusterPartitionFetcher.getInstance().updateRegionCache(req);
+    if (!result) {
+      return RpcUtils.getStatus(TSStatusCode.PARTITION_CACHE_UPDATE_ERROR);
+    }
+
+    if (SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
       // Notify consensus subscription queues of any preferred-writer changes
       try {
         ConsensusSubscriptionSetupHandler.onRegionRouteChanged(
@@ -2449,10 +2466,8 @@ public TSStatus updateRegionCache(TRegionRouteReq req) {
                 .MISC_LOG_FAILED_TO_PROCESS_CONSENSUS_SUBSCRIPTION_ROUTE_UPDATE_80D73E2B,
             e);
       }
-      return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
-    } else {
-      return RpcUtils.getStatus(TSStatusCode.PARTITION_CACHE_UPDATE_ERROR);
     }
+    return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
   }
 
   private Map<TConsensusGroupId, Boolean> getJudgedLeaders() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java

@@ -51,6 +51,7 @@
 import org.apache.iotdb.commons.service.RegisterManager;
 import org.apache.iotdb.commons.service.ServiceType;
 import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
 import org.apache.iotdb.commons.trigger.TriggerInformation;
 import org.apache.iotdb.commons.trigger.exception.TriggerManagementException;
 import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
@@ -941,7 +942,9 @@ private void setUp() throws StartupException, IOException {
     registerInternalRPCService();
 
     // Register subscription agent before pipe agent
-    registerManager.register(SubscriptionAgent.runtime());
+    if (SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      registerManager.register(SubscriptionAgent.runtime());
+    }
     registerManager.register(PipeDataNodeAgent.runtime());
 
     // Start GRASS Service

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

@@ -75,7 +75,7 @@ public class SubscriptionReceiverAgent {
   private final ScheduledExecutorService receiverTimeoutChecker;
 
   SubscriptionReceiverAgent() {
-    this(SubscriptionReceiverV1::new, true);
+    this(SubscriptionReceiverV1::new, SubscriptionConfig.getInstance().getSubscriptionEnabled());
   }
 
   SubscriptionReceiverAgent(

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

@@ -28,6 +28,7 @@
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.consensus.IConsensus;
 import org.apache.iotdb.consensus.iot.IoTConsensus;
@@ -741,6 +742,10 @@ public static void handleNewSubscriptions(
 
   public static void applyRuntimeState(
       final TConsensusGroupId groupId, final ConsensusRegionRuntimeState runtimeState) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return;
+    }
+
     final int newPreferredNodeId = runtimeState.getPreferredWriterNodeId();
     final Integer oldPreferredBoxed = lastKnownPreferredWriter.put(groupId, newPreferredNodeId);
     final int oldPreferredNodeId = (oldPreferredBoxed != null) ? oldPreferredBoxed : -1;
@@ -772,6 +777,10 @@ public static void applyRuntimeState(
 
   public static void onRegionRouteChanged(
       final Map<TConsensusGroupId, TRegionReplicaSet> newMap, final long routingTimestamp) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return;
+    }
+
     final int myNodeId = IOTDB_CONFIG.getDataNodeId();
 
     for (final Map.Entry<TConsensusGroupId, TRegionReplicaSet> newEntry : newMap.entrySet()) {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java

@@ -0,0 +1,64 @@
+/*
+ * 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.db.protocol.thrift.impl;
+
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.service.DataNode.DataNodeContext;
+import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class DataNodeInternalRPCServiceImplSubscriptionDisabledTest {
+
+  @BeforeClass
+  public static void setUp() {
+    IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
+  }
+
+  @Test
+  public void testSubscriptionRuntimeRPCsAreNoOpWhenSubscriptionIsDisabled() {
+    final boolean subscriptionEnabled =
+        CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
+    try {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false);
+      final DataNodeInternalRPCServiceImpl service =
+          new DataNodeInternalRPCServiceImpl(Mockito.mock(DataNodeContext.class));
+
+      final TPullCommitProgressResp pullResp = service.pullCommitProgress(null);
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), pullResp.getStatus().getCode());
+      Assert.assertTrue(pullResp.isSetCommitRegionProgress());
+      Assert.assertTrue(pullResp.getCommitRegionProgress().isEmpty());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          service.syncSubscriptionProgress(null).getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          service.pushSubscriptionRuntime(null).getCode());
+    } finally {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
+    }
+  }
+}

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.subscription.agent;
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.db.subscription.receiver.SubscriptionReceiver;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
@@ -38,17 +39,56 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
 public class SubscriptionReceiverAgentTest {
 
+  @Test
+  public void testTimeoutCheckerIsNotScheduledWhenSubscriptionIsDisabled() throws Exception {
+    final boolean subscriptionEnabled =
+        CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
+    try {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false);
+
+      final SubscriptionReceiverAgent agent = new SubscriptionReceiverAgent();
+
+      Assert.assertNull(getReceiverTimeoutChecker(agent));
+    } finally {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
+    }
+  }
+
+  @Test
+  public void testTimeoutCheckerIsScheduledWhenSubscriptionIsEnabled() throws Exception {
+    final boolean subscriptionEnabled =
+        CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
+    SubscriptionReceiverAgent agent = null;
+    try {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(true);
+
+      agent = new SubscriptionReceiverAgent();
+
+      Assert.assertNotNull(getReceiverTimeoutChecker(agent));
+    } finally {
+      if (agent != null) {
+        final ScheduledExecutorService receiverTimeoutChecker = getReceiverTimeoutChecker(agent);
+        if (receiverTimeoutChecker != null) {
+          receiverTimeoutChecker.shutdownNow();
+        }
+      }
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
+    }
+  }
+
   @Test
   public void testDisconnectedReceiverIsRetainedUntilTimeout() throws IOException {
     final CopyOnWriteArrayList<FakeSubscriptionReceiver> receivers = new CopyOnWriteArrayList<>();
@@ -176,6 +216,13 @@ private SubscriptionReceiverAgent createAgent(
     return new SubscriptionReceiverAgent(constructor, false);
   }
 
+  private ScheduledExecutorService getReceiverTimeoutChecker(final SubscriptionReceiverAgent agent)
+      throws Exception {
+    final Field field = SubscriptionReceiverAgent.class.getDeclaredField("receiverTimeoutChecker");
+    field.setAccessible(true);
+    return (ScheduledExecutorService) field.get(agent);
+  }
+
   private TPipeSubscribeReq createHandshakeRequest(
       final String consumerGroupId, final String consumerId) throws IOException {
     final Map<String, String> attributes = new HashMap<>();

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.subscription.broker.consensus;
 
+import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.consensus.DataRegionId;
 import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -49,6 +50,20 @@ public class ConsensusSubscriptionSetupHandlerTest {
 
   @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
 
+  @Test
+  public void testRuntimeUpdatesAreIgnoredWhenSubscriptionIsDisabled() {
+    final boolean subscriptionEnabled =
+        CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
+    try {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false);
+
+      ConsensusSubscriptionSetupHandler.applyRuntimeState(null, null);
+      ConsensusSubscriptionSetupHandler.onRegionRouteChanged(null, 0);
+    } finally {
+      CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
+    }
+  }
+
   @Test
   public void testSingleTopicSetupFailurePropagates() {
     SubscriptionException failure = null;