[Subscription] Disable subscription by default

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

@@ -77,6 +77,7 @@ public void setUp() throws Exception {
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()
+        .setSubscriptionEnabled(true)
         .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
         .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
         .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)

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

@@ -38,6 +38,7 @@ public void setUp() throws Exception {
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()
+        .setSubscriptionEnabled(true)
         .setAutoCreateSchemaEnabled(true)
         .setPipeMemoryManagementEnabled(false)
         .setIsPipeEnableMemoryCheck(false);

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

@@ -49,9 +49,17 @@ public void setUp() throws Exception {
   protected void setUpConfig() {
     senderEnv.getConfig().getDataNodeConfig().setDataNodeMemoryProportion("3:3:1:1:3:1");
 
-    // enable auto create schema
-    senderEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
-    receiverEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
+    // enable subscription and auto create schema
+    senderEnv
+        .getConfig()
+        .getCommonConfig()
+        .setSubscriptionEnabled(true)
+        .setAutoCreateSchemaEnabled(true);
+    receiverEnv
+        .getConfig()
+        .getCommonConfig()
+        .setSubscriptionEnabled(true)
+        .setAutoCreateSchemaEnabled(true);
 
     // 10 min, assert that the operations will not time out
     senderEnv.getConfig().getCommonConfig().setDnConnectionTimeoutMs(600000);

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

@@ -55,10 +55,22 @@ public void setUp() throws Exception {
   }
 
   protected void setUpConfig() {
-    // enable auto create schema
-    sender.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
-    receiver1.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
-    receiver2.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
+    // enable subscription and auto create schema
+    sender
+        .getConfig()
+        .getCommonConfig()
+        .setSubscriptionEnabled(true)
+        .setAutoCreateSchemaEnabled(true);
+    receiver1
+        .getConfig()
+        .getCommonConfig()
+        .setSubscriptionEnabled(true)
+        .setAutoCreateSchemaEnabled(true);
+    receiver2
+        .getConfig()
+        .getCommonConfig()
+        .setSubscriptionEnabled(true)
+        .setAutoCreateSchemaEnabled(true);
 
     // 10 min, assert that the operations will not time out
     sender.getConfig().getCommonConfig().setDnConnectionTimeoutMs(600000);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java

@@ -558,7 +558,7 @@ private void initQueryEngineMemoryAllocate(
         properties.getProperty("chunk_timeseriesmeta_free_memory_proportion");
     boolean subscriptionEnabled =
         Boolean.parseBoolean(
-            properties.getProperty("subscription_enabled", Boolean.TRUE.toString()));
+            properties.getProperty("subscription_enabled", Boolean.FALSE.toString()));
     final int[] queryMemoryProportions;
     try {
       queryMemoryProportions =

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java

@@ -21,10 +21,12 @@
 
 import org.apache.iotdb.commons.conf.TrimProperties;
 import org.apache.iotdb.commons.memory.MemoryConfig;
+import org.apache.iotdb.commons.memory.MemoryManager;
 import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
 
 import org.junit.Test;
 
+import java.lang.reflect.Method;
 import java.util.Arrays;
 
 import static org.junit.Assert.assertArrayEquals;
@@ -35,18 +37,60 @@
 public class DataNodeMemoryConfigTest {
 
   @Test
-  public void testResolveSubscriptionQueryMemoryProportions() {
+  public void testResolveSubscriptionQueryMemoryProportionsWhenEnabled() {
     final int[] defaultProportions = DataNodeMemoryConfig.resolveQueryMemoryProportions(null, true);
     assertArrayEquals(new int[] {1, 100, 200, 50, 200, 200, 200, 50, 250}, defaultProportions);
     assertEquals(
         0.2, (double) defaultProportions[8] / Arrays.stream(defaultProportions).sum(), 0.001);
     assertArrayEquals(
         new int[] {1, 100, 200, 50, 200, 200, 200, 50, 250},
         DataNodeMemoryConfig.resolveQueryMemoryProportions("1:100:200:50:200:200:200:50", true));
+  }
+
+  @Test
+  public void testResolveSubscriptionQueryMemoryProportionsWhenDisabled() {
     assertArrayEquals(
         new int[] {1, 100, 200, 50, 200, 200, 200, 50, 0},
         DataNodeMemoryConfig.resolveQueryMemoryProportions(
             "1:100:200:50:200:200:200:50:1000", false));
+    assertArrayEquals(
+        new int[] {1, 100, 200, 50, 200, 200, 200, 50, 0},
+        DataNodeMemoryConfig.resolveQueryMemoryProportions(null, false));
+  }
+
+  @Test
+  public void testSubscriptionDoesNotReserveQueryMemoryWhenDisabledByDefault()
+      throws ReflectiveOperationException {
+    final TrimProperties properties = new TrimProperties();
+    properties.setProperty("chunk_timeseriesmeta_free_memory_proportion", "0:0:0:0:1:0:0:0:1");
+    final DataNodeMemoryConfig memoryConfig = initializeQueryEngineMemory(properties);
+
+    assertEquals(0, memoryConfig.getSubscriptionMemoryManager().getTotalMemorySizeInBytes());
+    assertEquals(1_000_000L, memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes());
+  }
+
+  @Test
+  public void testSubscriptionDoesNotReserveQueryMemoryWhenExplicitlyDisabled()
+      throws ReflectiveOperationException {
+    final TrimProperties properties = new TrimProperties();
+    properties.setProperty("chunk_timeseriesmeta_free_memory_proportion", "0:0:0:0:1:0:0:0:1");
+    properties.setProperty("subscription_enabled", Boolean.FALSE.toString());
+    final DataNodeMemoryConfig memoryConfig = initializeQueryEngineMemory(properties);
+
+    assertEquals(0, memoryConfig.getSubscriptionMemoryManager().getTotalMemorySizeInBytes());
+    assertEquals(1_000_000L, memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes());
+  }
+
+  @Test
+  public void testSubscriptionReservesQueryMemoryWhenExplicitlyEnabled()
+      throws ReflectiveOperationException {
+    final TrimProperties properties = new TrimProperties();
+    properties.setProperty("chunk_timeseriesmeta_free_memory_proportion", "0:0:0:0:1:0:0:0:1");
+    properties.setProperty("subscription_enabled", Boolean.TRUE.toString());
+    final DataNodeMemoryConfig memoryConfig = initializeQueryEngineMemory(properties);
+
+    assertEquals(500_000L, memoryConfig.getSubscriptionMemoryManager().getTotalMemorySizeInBytes());
+    assertEquals(500_000L, memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes());
   }
 
   @Test
@@ -133,4 +177,15 @@ public void testCalculateAutoResizingBufferMemorySizeWithDeprecatedMemoryProport
         Runtime.getRuntime().maxMemory() / 7,
         DataNodeMemoryConfig.calculateAutoResizingBufferMemorySizeInBytes(properties));
   }
+
+  private DataNodeMemoryConfig initializeQueryEngineMemory(TrimProperties properties)
+      throws ReflectiveOperationException {
+    final DataNodeMemoryConfig memoryConfig = new DataNodeMemoryConfig();
+    final Method initQueryEngineMemoryAllocate =
+        DataNodeMemoryConfig.class.getDeclaredMethod(
+            "initQueryEngineMemoryAllocate", MemoryManager.class, TrimProperties.class);
+    initQueryEngineMemoryAllocate.setAccessible(true);
+    initQueryEngineMemoryAllocate.invoke(memoryConfig, new MemoryManager(1_000_000L), properties);
+    return memoryConfig;
+  }
 }

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;
@@ -34,7 +35,9 @@
 import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeReq;
 import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeResp;
 
+import org.junit.After;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 
 import java.io.IOException;
@@ -49,6 +52,20 @@
 
 public class SubscriptionReceiverAgentTest {
 
+  private boolean originalSubscriptionEnabled;
+
+  @Before
+  public void setUp() {
+    originalSubscriptionEnabled =
+        CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
+    CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(true);
+  }
+
+  @After
+  public void tearDown() {
+    CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(originalSubscriptionEnabled);
+  }
+
   @Test
   public void testDisconnectedReceiverIsRetainedUntilTimeout() throws IOException {
     final CopyOnWriteArrayList<FakeSubscriptionReceiver> receivers = new CopyOnWriteArrayList<>();

iotdb-core/datanode/src/test/resources/iotdb-system.properties

@@ -35,6 +35,9 @@ load_active_listening_dirs=target/ext/load/pending
 load_active_listening_pipe_dir=target/ext/load/pipe
 load_active_listening_fail_dir=target/ext/load/failed
 
+# Enable subscription for DataNode unit tests.
+subscription_enabled=true
+
 ####################
 ### REST Service Configuration
 ####################

iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template

@@ -2087,6 +2087,16 @@ pipe_air_gap_receiver_port=9780
 # Datatype: double
 pipe_all_sinks_rate_limit_bytes_per_second=-1
 
+####################
+### Subscription Configuration
+####################
+
+# Whether to enable subscription.
+# When disabled, subscription does not consume query memory.
+# effectiveMode: restart
+# Datatype: boolean
+subscription_enabled=false
+
 ####################
 ### Subscription Consensus Configuration
 ####################

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

@@ -386,7 +386,7 @@ public class CommonConfig {
 
   private boolean pipeAutoSplitFullEnabled = true;
 
-  private boolean subscriptionEnabled = true;
+  private boolean subscriptionEnabled = false;
 
   private float subscriptionCacheMemoryUsagePercentage = 0.2F;
   private int subscriptionSubtaskExecutorMaxThreadNum = 2;

iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java

@@ -0,0 +1,42 @@
+/*
+ * 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.commons.conf;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class CommonConfigTest {
+
+  @Test
+  public void testSubscriptionDisabledByDefaultInCommonConfig() {
+    assertFalse(new CommonConfig().getSubscriptionEnabled());
+  }
+
+  @Test
+  public void testSubscriptionDisabledByDefaultInConfigurationTemplate() throws IOException {
+    assertEquals(
+        Boolean.FALSE.toString(),
+        ConfigurationFileUtils.getConfigurationDefaultValue("subscription_enabled"));
+  }
+}