From c29d8a9e085d021641591853b7382c38bf562cd2 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 12 Aug 2026 18:51:23 +0800 Subject: [PATCH 1/2] [Subscription] Disable subscription by default --- .../cluster/IoTDBSubscriptionRestartIT.java | 1 + .../AbstractSubscriptionConsensusLocalIT.java | 1 + .../it/dual/AbstractSubscriptionDualIT.java | 14 ++++- .../triple/AbstractSubscriptionTripleIT.java | 20 +++++-- .../iotdb/db/conf/DataNodeMemoryConfig.java | 2 +- .../db/conf/DataNodeMemoryConfigTest.java | 57 ++++++++++++++++++- .../conf/iotdb-system.properties.template | 10 ++++ .../iotdb/commons/conf/CommonConfig.java | 2 +- .../iotdb/commons/conf/CommonConfigTest.java | 42 ++++++++++++++ 9 files changed, 139 insertions(+), 10 deletions(-) create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java index 44afe26da9ab..9022740bb09c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java +++ b/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) diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/AbstractSubscriptionConsensusLocalIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/AbstractSubscriptionConsensusLocalIT.java index 4342918c2bed..514258a16ea1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/AbstractSubscriptionConsensusLocalIT.java +++ b/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); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java index 45b6547422c9..f91414781697 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java +++ b/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); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/AbstractSubscriptionTripleIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/AbstractSubscriptionTripleIT.java index 7ffbfdf76bda..414f3b6bcfa8 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/AbstractSubscriptionTripleIT.java +++ b/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); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java index c63cd2c7ee8c..b021f19eaac6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java +++ b/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 = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java index 6c3e1b837129..896a1afbbd00 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java +++ b/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,7 +37,7 @@ 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( @@ -43,10 +45,52 @@ public void testResolveSubscriptionQueryMemoryProportions() { 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; + } } diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index c4a06b68b149..3b2de9daf5bd 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/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 #################### diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index 4adacb0d7aa1..d278321deefb 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/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; diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java new file mode 100644 index 000000000000..8ed00621e058 --- /dev/null +++ b/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")); + } +} From f084bc08bef2b1d455dcb52e430fa944103a6cb5 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 14 Aug 2026 12:17:56 +0800 Subject: [PATCH 2/2] [Subscription] Enable subscription in DataNode unit tests --- .../agent/SubscriptionReceiverAgentTest.java | 17 +++++++++++++++++ .../src/test/resources/iotdb-system.properties | 3 +++ 2 files changed, 20 insertions(+) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java index a151e83e21d0..0ba5835d6099 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java +++ b/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 receivers = new CopyOnWriteArrayList<>(); diff --git a/iotdb-core/datanode/src/test/resources/iotdb-system.properties b/iotdb-core/datanode/src/test/resources/iotdb-system.properties index 9e0e16caaa99..04e546c902ea 100644 --- a/iotdb-core/datanode/src/test/resources/iotdb-system.properties +++ b/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 ####################