Subscription: add config to enable or disable subscription

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java

@@ -526,6 +526,12 @@ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
     return this;
   }
 
+  @Override
+  public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
+    setProperty("subscription_enabled", String.valueOf(subscriptionEnabled));
+    return this;
+  }
+
   @Override
   public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
     setProperty("default_storage_group_level", String.valueOf(defaultStorageGroupLevel));

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java

@@ -541,6 +541,13 @@ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
     return this;
   }
 
+  @Override
+  public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
+    dnConfig.setSubscriptionEnabled(subscriptionEnabled);
+    cnConfig.setSubscriptionEnabled(subscriptionEnabled);
+    return this;
+  }
+
   @Override
   public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
     dnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel);

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java

@@ -380,4 +380,9 @@ public CommonConfig setPipeConnectorRequestSliceThresholdBytes(
   public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
     return this;
   }
+
+  @Override
+  public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java

@@ -168,6 +168,8 @@ CommonConfig setPipeConnectorRequestSliceThresholdBytes(
 
   CommonConfig setQueryMemoryProportion(String queryMemoryProportion);
 
+  CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled);
+
   default CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
     return this;
   }

integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java

@@ -56,6 +56,8 @@ public class IoTDBDatabaseIT {
 
   @Before
   public void setUp() throws Exception {
+    // enable subscription
+    EnvFactory.getEnv().getConfig().getCommonConfig().setSubscriptionEnabled(true);
     EnvFactory.getEnv().initClusterEnvironment();
   }
 

integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java

@@ -71,6 +71,10 @@ public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT {
   public void setUp() throws Exception {
     super.setUp();
 
+    // enable subscription
+    EnvFactory.getEnv().getConfig().getCommonConfig().setSubscriptionEnabled(true);
+
+    // set cluster env
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()

integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java

@@ -47,6 +47,10 @@ public void setUp() throws Exception {
   }
 
   protected void setUpConfig() {
+    // enable subscription
+    senderEnv.getConfig().getCommonConfig().setSubscriptionEnabled(true);
+    receiverEnv.getConfig().getCommonConfig().setSubscriptionEnabled(true);
+
     // enable auto create schema
     senderEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
     receiverEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);

integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java

@@ -32,6 +32,8 @@ public abstract class AbstractSubscriptionLocalIT extends AbstractSubscriptionIT
   public void setUp() throws Exception {
     super.setUp();
 
+    // enable subscription
+    EnvFactory.getEnv().getConfig().getCommonConfig().setSubscriptionEnabled(true);
     EnvFactory.getEnv().initClusterEnvironment();
   }
 

integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/AbstractSubscriptionTripleIT.java

@@ -55,6 +55,11 @@ public void setUp() throws Exception {
   }
 
   protected void setUpConfig() {
+    // enable subscription
+    sender.getConfig().getCommonConfig().setSubscriptionEnabled(true);
+    receiver1.getConfig().getCommonConfig().setSubscriptionEnabled(true);
+    receiver2.getConfig().getCommonConfig().setSubscriptionEnabled(true);
+
     // enable auto create schema
     sender.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
     receiver1.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);

integration-test/src/test/java/org/apache/iotdb/tools/it/ExportTsFileTestIT.java

@@ -52,7 +52,10 @@ public class ExportTsFileTestIT extends AbstractScriptIT {
 
   @BeforeClass
   public static void setUp() throws Exception {
+    // enable subscription
+    EnvFactory.getEnv().getConfig().getCommonConfig().setSubscriptionEnabled(true);
     EnvFactory.getEnv().initClusterEnvironment();
+
     ip = EnvFactory.getEnv().getIP();
     port = EnvFactory.getEnv().getPort();
     toolsPath = EnvFactory.getEnv().getToolsPath();

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -295,6 +295,7 @@ public enum TSStatusCode {
   SUBSCRIPTION_MISSING_CUSTOMER(1909),
   SHOW_SUBSCRIPTION_ERROR(1910),
   SUBSCRIPTION_PIPE_TIMEOUT_ERROR(1911),
+  SUBSCRIPTION_NOT_ENABLED_ERROR(1912),
 
   // Topic
   CREATE_TOPIC_ERROR(2000),

iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProvider.java

@@ -432,6 +432,7 @@ private static void verifyPipeSubscribeSuccess(final TSStatus status)
       case 1900: // SUBSCRIPTION_VERSION_ERROR
       case 1901: // SUBSCRIPTION_TYPE_ERROR
       case 1909: // SUBSCRIPTION_MISSING_CUSTOMER
+      case 1912: // SUBSCRIPTION_NOT_ENABLED_ERROR
       default:
         {
           final String errorMessage =

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

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.pipe.agent.task.execution;
 
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
 import org.apache.iotdb.db.pipe.consensus.PipeConsensusSubtaskExecutor;
 import org.apache.iotdb.db.subscription.task.execution.SubscriptionSubtaskExecutor;
 
@@ -53,7 +54,10 @@ public PipeConsensusSubtaskExecutor getConsensusExecutor() {
   private PipeSubtaskExecutorManager() {
     processorExecutor = new PipeProcessorSubtaskExecutor();
     connectorExecutor = new PipeConnectorSubtaskExecutor();
-    subscriptionExecutor = new SubscriptionSubtaskExecutor();
+    subscriptionExecutor =
+        SubscriptionConfig.getInstance().getSubscriptionEnabled()
+            ? new SubscriptionSubtaskExecutor()
+            : null;
     consensusExecutor = new PipeConsensusSubtaskExecutor();
   }
 

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

@@ -68,6 +68,7 @@
 import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
 import org.apache.iotdb.commons.service.metric.MetricService;
 import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
 import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
 import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
 import org.apache.iotdb.commons.trigger.TriggerInformation;
@@ -1239,6 +1240,11 @@ public TPushPipeMetaResp pushMultiPipeMeta(TPushMultiPipeMetaReq req) {
 
   @Override
   public TPushTopicMetaResp pushTopicMeta(TPushTopicMetaReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TPushTopicMetaResp()
+          .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+    }
+
     final List<TopicMeta> topicMetas = new ArrayList<>();
     for (ByteBuffer byteBuffer : req.getTopicMetas()) {
       topicMetas.add(TopicMeta.deserialize(byteBuffer));
@@ -1262,6 +1268,11 @@ public TPushTopicMetaResp pushTopicMeta(TPushTopicMetaReq req) {
 
   @Override
   public TPushTopicMetaResp pushSingleTopicMeta(TPushSingleTopicMetaReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TPushTopicMetaResp()
+          .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+    }
+
     try {
       final TPushTopicMetaRespExceptionMessage exceptionMessage;
       if (req.isSetTopicNameToDrop()) {
@@ -1290,6 +1301,11 @@ public TPushTopicMetaResp pushSingleTopicMeta(TPushSingleTopicMetaReq req) {
 
   @Override
   public TPushTopicMetaResp pushMultiTopicMeta(TPushMultiTopicMetaReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TPushTopicMetaResp()
+          .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+    }
+
     boolean hasException = false;
     // If there is any exception, we use the size of exceptionMessages to record the fail index
     List<TPushTopicMetaRespExceptionMessage> exceptionMessages = new ArrayList<>();
@@ -1337,6 +1353,11 @@ public TPushTopicMetaResp pushMultiTopicMeta(TPushMultiTopicMetaReq req) {
 
   @Override
   public TPushConsumerGroupMetaResp pushConsumerGroupMeta(TPushConsumerGroupMetaReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TPushConsumerGroupMetaResp()
+          .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+    }
+
     final List<ConsumerGroupMeta> consumerGroupMetas = new ArrayList<>();
     for (ByteBuffer byteBuffer : req.getConsumerGroupMetas()) {
       consumerGroupMetas.add(ConsumerGroupMeta.deserialize(byteBuffer));
@@ -1361,6 +1382,11 @@ public TPushConsumerGroupMetaResp pushConsumerGroupMeta(TPushConsumerGroupMetaRe
   @Override
   public TPushConsumerGroupMetaResp pushSingleConsumerGroupMeta(
       TPushSingleConsumerGroupMetaReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return new TPushConsumerGroupMetaResp()
+          .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+    }
+
     try {
       final TPushConsumerGroupMetaRespExceptionMessage exceptionMessage;
       if (req.isSetConsumerGroupNameToDrop()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java

@@ -66,6 +66,7 @@
 import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchemaUtil;
 import org.apache.iotdb.commons.schema.view.LogicalViewSchema;
 import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
 import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
 import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
 import org.apache.iotdb.commons.udf.service.UDFClassLoader;
@@ -375,6 +376,16 @@ private static final class ClusterConfigTaskExecutorHolder {
     private ClusterConfigTaskExecutorHolder() {}
   }
 
+  private static final SettableFuture<ConfigTaskResult> SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+
+  static {
+    SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE = SettableFuture.create();
+    SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE.setException(
+        new IoTDBException(
+            "Subscription not enabled, please set config `subscription_enabled` to true.",
+            TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR.getStatusCode()));
+  }
+
   public static ClusterConfigTaskExecutor getInstance() {
     return ClusterConfigTaskExecutor.ClusterConfigTaskExecutorHolder.INSTANCE;
   }
@@ -2391,6 +2402,10 @@ public SettableFuture<ConfigTaskResult> showPipes(final ShowPipesStatement showP
   @Override
   public SettableFuture<ConfigTaskResult> showSubscriptions(
       final ShowSubscriptionsStatement showSubscriptionsStatement) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+    }
+
     final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
 
     try (final ConfigNodeClient configNodeClient =
@@ -2426,6 +2441,10 @@ public SettableFuture<ConfigTaskResult> showSubscriptions(
   @Override
   public SettableFuture<ConfigTaskResult> dropSubscription(
       final DropSubscriptionStatement dropSubscriptionStatement) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+    }
+
     final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
     try (ConfigNodeClient configNodeClient =
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
@@ -2449,6 +2468,10 @@ public SettableFuture<ConfigTaskResult> dropSubscription(
   @Override
   public SettableFuture<ConfigTaskResult> createTopic(
       final CreateTopicStatement createTopicStatement) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+    }
+
     final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
 
     final String topicName = createTopicStatement.getTopicName();
@@ -2515,6 +2538,10 @@ public SettableFuture<ConfigTaskResult> createTopic(
 
   @Override
   public SettableFuture<ConfigTaskResult> dropTopic(final DropTopicStatement dropTopicStatement) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+    }
+
     final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
     try (ConfigNodeClient configNodeClient =
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
@@ -2538,6 +2565,10 @@ public SettableFuture<ConfigTaskResult> dropTopic(final DropTopicStatement dropT
   @Override
   public SettableFuture<ConfigTaskResult> showTopics(
       final ShowTopicsStatement showTopicsStatement) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+    }
+
     final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
     try (final ConfigNodeClient configNodeClient =
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {

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

@@ -42,17 +42,29 @@ public class SubscriptionReceiverAgent {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionReceiverAgent.class);
 
-  private final ThreadLocal<SubscriptionReceiver> receiverThreadLocal = new ThreadLocal<>();
-
   private static final Map<Byte, Supplier<SubscriptionReceiver>> RECEIVER_CONSTRUCTORS =
       new HashMap<>();
 
+  private static final TPipeSubscribeResp SUBSCRIPTION_NOT_ENABLED_ERROR_RESP =
+      new TPipeSubscribeResp(
+          RpcUtils.getStatus(
+              TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR,
+              "Subscription not enabled, please set config `subscription_enabled` to true."),
+          PipeSubscribeResponseVersion.VERSION_1.getVersion(),
+          PipeSubscribeResponseType.ACK.getType());
+
+  private final ThreadLocal<SubscriptionReceiver> receiverThreadLocal = new ThreadLocal<>();
+
   SubscriptionReceiverAgent() {
     RECEIVER_CONSTRUCTORS.put(
         PipeSubscribeRequestVersion.VERSION_1.getVersion(), SubscriptionReceiverV1::new);
   }
 
   public TPipeSubscribeResp handle(final TPipeSubscribeReq req) {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return SUBSCRIPTION_NOT_ENABLED_ERROR_RESP;
+    }
+
     final byte reqVersion = req.getVersion();
     if (RECEIVER_CONSTRUCTORS.containsKey(reqVersion)) {
       return getReceiver(reqVersion).handle(req);

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

@@ -62,6 +62,10 @@ SubscriptionTopicAgent topic() {
 
   @Override
   public void start() throws StartupException {
+    if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return;
+    }
+
     SubscriptionConfig.getInstance().printAllConfigs();
 
     SubscriptionAgentLauncher.launchSubscriptionTopicAgent();

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java

@@ -324,6 +324,8 @@ public class CommonConfig {
   private boolean pipeEventReferenceTrackingEnabled = true;
   private long pipeEventReferenceEliminateIntervalSeconds = 10;
 
+  private boolean subscriptionEnabled = false;
+
   private float subscriptionCacheMemoryUsagePercentage = 0.2F;
   private int subscriptionSubtaskExecutorMaxThreadNum = 2;
 
@@ -2156,6 +2158,14 @@ public void setPipeEventReferenceEliminateIntervalSeconds(
         pipeEventReferenceEliminateIntervalSeconds);
   }
 
+  public boolean getSubscriptionEnabled() {
+    return subscriptionEnabled;
+  }
+
+  public void setSubscriptionEnabled(boolean subscriptionEnabled) {
+    this.subscriptionEnabled = subscriptionEnabled;
+  }
+
   public float getSubscriptionCacheMemoryUsagePercentage() {
     return subscriptionCacheMemoryUsagePercentage;
   }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java

@@ -256,6 +256,11 @@ public void loadCommonProps(TrimProperties properties) throws IOException {
   }
 
   private void loadSubscriptionProps(TrimProperties properties) {
+    config.setSubscriptionEnabled(
+        Boolean.parseBoolean(
+            properties.getProperty(
+                "subscription_enabled", String.valueOf(config.getSubscriptionEnabled()))));
+
     config.setSubscriptionCacheMemoryUsagePercentage(
         Float.parseFloat(
             properties.getProperty(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java

@@ -33,6 +33,7 @@
 import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
 import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
 import org.apache.iotdb.commons.pipe.connector.limiter.PipeEndPointRateLimiter;
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
 import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatReq;
 import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatResp;
 import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
@@ -167,6 +168,13 @@ protected TPushPipeMetaRespExceptionMessage handleSinglePipeMetaChangesInternal(
   private void executeSinglePipeMetaChanges(final PipeMeta metaFromCoordinator)
       throws IllegalPathException {
     final String pipeName = metaFromCoordinator.getStaticMeta().getPipeName();
+
+    // Do nothing with the subscription pipe if disable subscription
+    if (PipeStaticMeta.isSubscriptionPipe(pipeName)
+        && !SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+      return;
+    }
+
     final PipeMeta metaInAgent = pipeMetaKeeper.getPipeMeta(pipeName);
 
     // If pipe meta does not exist on local agent, create a new pipe