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/4] [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 44afe26da9ab4..9022740bb09cf 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 4342918c2bed8..514258a16ea11 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 45b6547422c9d..f914147816971 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 7ffbfdf76bdac..414f3b6bcfa82 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 c63cd2c7ee8c7..b021f19eaac6e 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 6c3e1b837129a..896a1afbbd009 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 c4a06b68b1497..3b2de9daf5bd4 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 4adacb0d7aa1c..d278321deefba 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 0000000000000..8ed00621e0589 --- /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/4] [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 a151e83e21d04..0ba5835d60995 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 9e0e16caaa99f..04e546c902eab 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 #################### From 45f443047767cd16ff41c710890d1970be14d28d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 24 Aug 2026 11:49:13 +0800 Subject: [PATCH 3/4] [Subscription] Refine subscription configuration --- .github/workflows/daily-it.yml | 359 ------------------ .github/workflows/pipe-it.yml | 350 ----------------- integration-test/pom.xml | 4 + .../env/cluster/config/MppCommonConfig.java | 6 - .../cluster/config/MppSharedCommonConfig.java | 7 - .../env/remote/config/RemoteCommonConfig.java | 5 - .../apache/iotdb/itbase/env/CommonConfig.java | 2 - .../cluster/IoTDBSubscriptionRestartIT.java | 1 - ...oTDBSubscriptionTopicOwnerPartitionIT.java | 1 - .../AbstractSubscriptionConsensusLocalIT.java | 1 - ...nsusSubscriptionColumnFilterClusterIT.java | 1 - .../it/dual/AbstractSubscriptionDualIT.java | 14 +- .../it/local/AbstractSubscriptionLocalIT.java | 2 - .../local/IoTDBSubscriptionTopicOwnerIT.java | 1 - .../triple/AbstractSubscriptionTripleIT.java | 20 +- .../iotdb/db/conf/DataNodeMemoryConfig.java | 5 +- .../agent/SubscriptionReceiverAgent.java | 16 +- .../db/conf/DataNodeMemoryConfigTest.java | 6 +- .../agent/SubscriptionReceiverAgentTest.java | 29 +- .../test/resources/iotdb-system.properties | 3 - .../conf/iotdb-system.properties.template | 10 - .../iotdb/commons/conf/CommonConfig.java | 9 +- .../iotdb/commons/conf/CommonDescriptor.java | 4 - .../iotdb/commons/conf/CommonConfigTest.java | 11 +- 24 files changed, 49 insertions(+), 818 deletions(-) diff --git a/.github/workflows/daily-it.yml b/.github/workflows/daily-it.yml index 745ffbd0769fe..553cf6f75145c 100644 --- a/.github/workflows/daily-it.yml +++ b/.github/workflows/daily-it.yml @@ -534,365 +534,6 @@ jobs: name: cluster-log-dual-tree-manual-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} path: integration-test/target/cluster-logs retention-days: 30 - SubscriptionTreeArchVerification: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: - [ - ScalableSingleNodeMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTreeArchVerification \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-tree-arch-verification-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - SubscriptionTableArchVerification: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: [ScalableSingleNodeMode] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTableArchVerification \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-table-arch-verification-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - SubscriptionTreeRegressionConsumer: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # do not use HighPerformanceMode here, otherwise some tests will cause the GH runner to receive a shutdown signal - cluster1: - [ - ScalableSingleNodeMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTreeRegressionConsumer \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-tree-regression-consumer-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - SubscriptionTreeRegressionMisc: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # do not use HighPerformanceMode here, otherwise some tests will cause the GH runner to receive a shutdown signal - cluster1: - [ - ScalableSingleNodeMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTreeRegressionMisc \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-tree-regression-misc-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 PipeDualTableManualBasic: strategy: fail-fast: false diff --git a/.github/workflows/pipe-it.yml b/.github/workflows/pipe-it.yml index 2a9fbbd12398b..7b5d6df0b9e11 100644 --- a/.github/workflows/pipe-it.yml +++ b/.github/workflows/pipe-it.yml @@ -451,356 +451,6 @@ jobs: name: cluster-log-dual-tree-manual-shard${{ matrix.shard }}-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} path: integration-test/target/cluster-logs retention-days: 30 - subscription-tree-arch-verification: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: [ScalableSingleNodeMode] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.clusterReadyRetryCount=90 \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTreeArchVerification \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-tree-arch-verification-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - subscription-table-arch-verification: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: [ScalableSingleNodeMode] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.clusterReadyRetryCount=90 \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTableArchVerification \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-table-arch-verification-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - subscription-tree-regression-consumer: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # do not use HighPerformanceMode here, otherwise some tests will cause the GH runner to receive a shutdown signal - cluster1: [ScalableSingleNodeMode] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.clusterReadyRetryCount=90 \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTreeRegressionConsumer \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-tree-regression-consumer-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - subscription-tree-regression-misc: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # do not use HighPerformanceMode here, otherwise some tests will cause the GH runner to receive a shutdown signal - cluster1: [ScalableSingleNodeMode] - cluster2: [ScalableSingleNodeMode] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@v5 - - name: Set up JDK ${{ matrix.java }} - uses: actions/setup-java@v5 - with: - distribution: corretto - java-version: ${{ matrix.java }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - name: Cache Maven packages - uses: actions/cache@v5 - with: - path: ~/.m2 - key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} - restore-keys: ${{ runner.os }}-m2- - - name: Sleep for a random duration between 0 and 10000 milliseconds - run: | - sleep $(( $(( RANDOM % 10000 + 1 )) / 1000)) - - name: IT Test - shell: bash - # we do not compile client-cpp for saving time, it is tested in client.yml - # we can skip influxdb-protocol because it has been tested separately in influxdb-protocol.yml - run: | - retry() { - local -i max_attempts=3 - local -i attempt=1 - local -i retry_sleep=5 - local test_output - - while [ $attempt -le $max_attempts ]; do - mvn clean verify \ - -P with-integration-tests \ - -DskipUTs \ - -DintegrationTest.clusterReadyRetryCount=90 \ - -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \ - -DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \ - -pl integration-test \ - -am -PMultiClusterIT2SubscriptionTreeRegressionMisc \ - -ntp >> ~/run-tests-$attempt.log && return 0 - test_output=$(cat ~/run-tests-$attempt.log) - - echo "==================== BEGIN: ~/run-tests-$attempt.log ====================" - echo "$test_output" - echo "==================== END: ~/run-tests-$attempt.log ======================" - - if ! mv ~/run-tests-$attempt.log integration-test/target/cluster-logs/ 2>/dev/null; then - echo "Failed to move log file ~/run-tests-$attempt.log to integration-test/target/cluster-logs/. Skipping..." - fi - - if echo "$test_output" | grep -q "Could not transfer artifact"; then - if [ $attempt -lt $max_attempts ]; then - echo "Test failed with artifact transfer issue, attempt $attempt. Retrying in $retry_sleep seconds..." - sleep $retry_sleep - attempt=$((attempt + 1)) - else - echo "Test failed after $max_attempts attempts due to artifact transfer issue." - echo "Treating this as a success because the issue is likely transient." - return 0 - fi - elif [ $? -ne 0 ]; then - echo "Test failed with a different error." - return 1 - else - echo "Tests passed" - return 0 - fi - done - } - retry - - name: Upload Artifact - if: failure() - uses: actions/upload-artifact@v6 - with: - name: cluster-log-subscription-tree-regression-misc-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - # 13 IT classes split across 3 parallel shards to cut the historical ~63 min - # wall clock to ~22 min. See cluster-it-1c1d.yml for the shard pattern. dual-table-manual-basic: strategy: fail-fast: false diff --git a/integration-test/pom.xml b/integration-test/pom.xml index 4dfea25247802..2152a5e113475 100644 --- a/integration-test/pom.xml +++ b/integration-test/pom.xml @@ -324,6 +324,10 @@ ${integrationTest.includedGroups} ${integrationTest.excludedGroups} + + + **/subscription/it/** + false none ${integrationTest.forkCount} diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java index ec67e5f451bd6..31f26def31815 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java @@ -495,12 +495,6 @@ public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) return this; } - @Override - public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) { - setProperty("subscription_enabled", String.valueOf(subscriptionEnabled)); - return this; - } - @Override public CommonConfig setSubscriptionOwnerLeaseDurationMsMin( long subscriptionOwnerLeaseDurationMsMin) { diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java index 36d06aefe889d..da61b00e8b11b 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java @@ -512,13 +512,6 @@ public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) return this; } - @Override - public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) { - dnConfig.setSubscriptionEnabled(subscriptionEnabled); - cnConfig.setSubscriptionEnabled(subscriptionEnabled); - return this; - } - @Override public CommonConfig setSubscriptionOwnerLeaseDurationMsMin( long subscriptionOwnerLeaseDurationMsMin) { diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java index 752dcd009db0a..1eb7757da4ff6 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java @@ -363,11 +363,6 @@ public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) return this; } - @Override - public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) { - return this; - } - @Override public CommonConfig setSubscriptionOwnerLeaseDurationMsMin( long subscriptionOwnerLeaseDurationMsMin) { diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java index 0ad3c23af16fe..f5324f3610124 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java @@ -159,8 +159,6 @@ CommonConfig setEnableAutoLeaderBalanceForRatisConsensus( CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck); - CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled); - CommonConfig setSubscriptionOwnerLeaseDurationMsMin(long subscriptionOwnerLeaseDurationMsMin); CommonConfig setPipeAirGapReceiverEnabled(boolean isPipeAirGapReceiverEnabled); 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 9022740bb09cf..44afe26da9ab4 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,7 +77,6 @@ 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/cluster/IoTDBSubscriptionTopicOwnerPartitionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionTopicOwnerPartitionIT.java index 31fa3fde02276..69ddc56297c78 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionTopicOwnerPartitionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionTopicOwnerPartitionIT.java @@ -68,7 +68,6 @@ public void setUp() throws Exception { EnvFactory.getEnv() .getConfig() .getCommonConfig() - .setSubscriptionEnabled(true) .setPipeMemoryManagementEnabled(false) .setIsPipeEnableMemoryCheck(false) // Lower the owner-lease floor so the test can use a short lease and stay fast. 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 514258a16ea11..4342918c2bed8 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,7 +38,6 @@ 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/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java index 6c33532828b09..e90d94af7ea7f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java @@ -63,7 +63,6 @@ public void setUp() throws Exception { .setSchemaReplicationFactor(1) .setDataReplicationFactor(2) .setAutoCreateSchemaEnabled(true) - .setSubscriptionEnabled(true) .setPipeMemoryManagementEnabled(false) .setIsPipeEnableMemoryCheck(false) .setSubscriptionOwnerLeaseDurationMsMin(1000); 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 f914147816971..45b6547422c9d 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,17 +49,9 @@ public void setUp() throws Exception { protected void setUpConfig() { senderEnv.getConfig().getDataNodeConfig().setDataNodeMemoryProportion("3:3:1:1:3:1"); - // enable subscription and auto create schema - senderEnv - .getConfig() - .getCommonConfig() - .setSubscriptionEnabled(true) - .setAutoCreateSchemaEnabled(true); - receiverEnv - .getConfig() - .getCommonConfig() - .setSubscriptionEnabled(true) - .setAutoCreateSchemaEnabled(true); + // enable auto create schema + senderEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true); + receiverEnv.getConfig().getCommonConfig().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/local/AbstractSubscriptionLocalIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java index fe667480c86bf..c177d03932715 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java @@ -32,11 +32,9 @@ public abstract class AbstractSubscriptionLocalIT extends AbstractSubscriptionIT public void setUp() throws Exception { super.setUp(); - // enable subscription EnvFactory.getEnv() .getConfig() .getCommonConfig() - .setSubscriptionEnabled(true) .setPipeMemoryManagementEnabled(false) .setIsPipeEnableMemoryCheck(false); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionTopicOwnerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionTopicOwnerIT.java index c08f151c16f49..9b506c049511f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionTopicOwnerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionTopicOwnerIT.java @@ -58,7 +58,6 @@ public void setUp() throws Exception { EnvFactory.getEnv() .getConfig() .getCommonConfig() - .setSubscriptionEnabled(true) .setPipeMemoryManagementEnabled(false) .setIsPipeEnableMemoryCheck(false) // Lower the owner-lease floor so the drain test can use a short lease and stay fast. 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 414f3b6bcfa82..7ffbfdf76bdac 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,22 +55,10 @@ public void setUp() throws Exception { } protected void setUpConfig() { - // 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); + // enable auto create schema + sender.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true); + receiver1.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true); + receiver2.getConfig().getCommonConfig().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 b021f19eaac6e..a6fe6e0ef9934 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.db.conf; +import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.ConfigurationFileUtils; import org.apache.iotdb.commons.conf.TrimProperties; import org.apache.iotdb.commons.memory.MemoryConfig; @@ -556,9 +557,7 @@ private void initQueryEngineMemoryAllocate( long maxMemoryAvailable = queryEngineMemoryManager.getTotalMemorySizeInBytes(); String queryMemoryAllocateProportion = properties.getProperty("chunk_timeseriesmeta_free_memory_proportion"); - boolean subscriptionEnabled = - Boolean.parseBoolean( - properties.getProperty("subscription_enabled", Boolean.FALSE.toString())); + boolean subscriptionEnabled = CommonConfig.SUBSCRIPTION_ENABLED; final int[] queryMemoryProportions; try { queryMemoryProportions = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java index d918b4aed474e..2b9b6dcd516f0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.subscription.receiver.SubscriptionReceiver; import org.apache.iotdb.db.subscription.receiver.SubscriptionReceiverV1; import org.apache.iotdb.rpc.RpcUtils; @@ -46,6 +47,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; import java.util.function.Supplier; public class SubscriptionReceiverAgent { @@ -56,7 +58,7 @@ public class SubscriptionReceiverAgent { new TPipeSubscribeResp( RpcUtils.getStatus( TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR, - "Subscription not enabled, please set config `subscription_enabled` to true."), + DataNodeQueryMessages.QUERY_EXCEPTION_SUBSCRIPTION_IS_NOT_ENABLED_7F43DCBB), PipeSubscribeResponseVersion.VERSION_1.getVersion(), PipeSubscribeResponseType.ACK.getType()); @@ -72,15 +74,21 @@ public class SubscriptionReceiverAgent { private final ConcurrentHashMap consumerReceivers = new ConcurrentHashMap<>(); + private final BooleanSupplier subscriptionEnabledSupplier; private final ScheduledExecutorService receiverTimeoutChecker; SubscriptionReceiverAgent() { - this(SubscriptionReceiverV1::new, true); + this( + SubscriptionReceiverV1::new, + true, + () -> SubscriptionConfig.getInstance().getSubscriptionEnabled()); } SubscriptionReceiverAgent( final Supplier receiverConstructor, - final boolean scheduleTimeoutChecker) { + final boolean scheduleTimeoutChecker, + final BooleanSupplier subscriptionEnabledSupplier) { + this.subscriptionEnabledSupplier = subscriptionEnabledSupplier; receiverConstructors.put( PipeSubscribeRequestVersion.VERSION_1.getVersion(), receiverConstructor); if (scheduleTimeoutChecker) { @@ -111,7 +119,7 @@ public TPipeSubscribeResp handle(final TPipeSubscribeReq req, final String usern PipeSubscribeResponseVersion.VERSION_1.getVersion(), PipeSubscribeResponseType.ACK.getType()); } - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!subscriptionEnabledSupplier.getAsBoolean()) { return SUBSCRIPTION_NOT_ENABLED_ERROR_RESP; } 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 896a1afbbd009..6aeeb22f2f642 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 @@ -82,15 +82,15 @@ public void testSubscriptionDoesNotReserveQueryMemoryWhenExplicitlyDisabled() } @Test - public void testSubscriptionReservesQueryMemoryWhenExplicitlyEnabled() + public void testSubscriptionDoesNotReserveQueryMemoryWhenConfiguredEnabled() 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()); + assertEquals(0, memoryConfig.getSubscriptionMemoryManager().getTotalMemorySizeInBytes()); + assertEquals(1_000_000L, memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes()); } @Test 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 0ba5835d60995..c381f7a3099cd 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,7 +20,6 @@ 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; @@ -35,9 +34,7 @@ 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; @@ -52,18 +49,20 @@ public class SubscriptionReceiverAgentTest { - private boolean originalSubscriptionEnabled; - - @Before - public void setUp() { - originalSubscriptionEnabled = - CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled(); - CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(true); - } + @Test + public void testDisabledSubscriptionRejectsRequest() throws IOException { + final SubscriptionReceiverAgent agent = + new SubscriptionReceiverAgent( + () -> { + throw new AssertionError( + "Receiver must not be created when subscription is disabled"); + }, + false, + () -> false); - @After - public void tearDown() { - CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(originalSubscriptionEnabled); + Assert.assertEquals( + TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR.getStatusCode(), + agent.handle(createHandshakeRequest("group", "consumer"), "root").getStatus().getCode()); } @Test @@ -190,7 +189,7 @@ private SubscriptionReceiverAgent createAgent( receivers.add(receiver); return receiver; }; - return new SubscriptionReceiverAgent(constructor, false); + return new SubscriptionReceiverAgent(constructor, false, () -> true); } private TPipeSubscribeReq createHandshakeRequest( diff --git a/iotdb-core/datanode/src/test/resources/iotdb-system.properties b/iotdb-core/datanode/src/test/resources/iotdb-system.properties index 04e546c902eab..9e0e16caaa99f 100644 --- a/iotdb-core/datanode/src/test/resources/iotdb-system.properties +++ b/iotdb-core/datanode/src/test/resources/iotdb-system.properties @@ -35,9 +35,6 @@ 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 #################### 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 3b2de9daf5bd4..c4a06b68b1497 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,16 +2087,6 @@ 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 d278321deefba..f4fca603dddf1 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 @@ -58,6 +58,7 @@ public class CommonConfig { public static final String SYSTEM_CONFIG_TEMPLATE_NAME = "iotdb-system.properties.template"; private static final Logger logger = LoggerFactory.getLogger(CommonConfig.class); public static final long DEFAULT_TIME_PARTITION_INTERVAL = 604_800_000L; + public static final boolean SUBSCRIPTION_ENABLED = false; // The authorizer provider class which extends BasicAuthorizer private String authorizerProvider = @@ -386,8 +387,6 @@ public class CommonConfig { private boolean pipeAutoSplitFullEnabled = true; - private boolean subscriptionEnabled = false; - private float subscriptionCacheMemoryUsagePercentage = 0.2F; private int subscriptionSubtaskExecutorMaxThreadNum = 2; private int subscriptionConsensusPrefetchExecutorMaxThreadNum = 2; @@ -2541,11 +2540,7 @@ public void setPipeAutoSplitFullEnabled(boolean pipeAutoSplitFullEnabled) { } public boolean getSubscriptionEnabled() { - return subscriptionEnabled; - } - - public void setSubscriptionEnabled(boolean subscriptionEnabled) { - this.subscriptionEnabled = subscriptionEnabled; + return SUBSCRIPTION_ENABLED; } public float getSubscriptionCacheMemoryUsagePercentage() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index 7de383a0d616d..b249dea07ee16 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java @@ -346,10 +346,6 @@ 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( 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 index 8ed00621e0589..60dd851754afd 100644 --- 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 @@ -23,20 +23,19 @@ import java.io.IOException; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; public class CommonConfigTest { @Test - public void testSubscriptionDisabledByDefaultInCommonConfig() { + public void testSubscriptionDisabledInCommonConfig() { + assertFalse(CommonConfig.SUBSCRIPTION_ENABLED); assertFalse(new CommonConfig().getSubscriptionEnabled()); } @Test - public void testSubscriptionDisabledByDefaultInConfigurationTemplate() throws IOException { - assertEquals( - Boolean.FALSE.toString(), - ConfigurationFileUtils.getConfigurationDefaultValue("subscription_enabled")); + public void testSubscriptionIsNotExposedInConfigurationTemplate() throws IOException { + assertNull(ConfigurationFileUtils.getConfigurationDefaultValue("subscription_enabled")); } } From 47fcb39d3280c7b4dacdcb6eeb4fd09e85c9dd45 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:31:49 +0800 Subject: [PATCH 4/4] Update pom.xml --- pom.xml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pom.xml b/pom.xml index b4ab0ad31332f..5755e72cfff89 100644 --- a/pom.xml +++ b/pom.xml @@ -722,6 +722,10 @@ 3.1.2 ${argLine} -Xmx1024m + + + **/subscription/** + ${project.build.directory}/fork_${surefire.forkNumber}