This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a8950536244 [Subscription] Disable subscription by default (#18457)
a8950536244 is described below
commit a8950536244f1cbe6a4c5060a5bd5ff62b7c2f66
Author: Caideyipi <[email protected]>
AuthorDate: Thu Aug 27 10:16:24 2026 +0800
[Subscription] Disable subscription by default (#18457)
* [Subscription] Disable subscription by default
* [Subscription] Enable subscription in DataNode unit tests
* [Subscription] Refine subscription configuration
* Update pom.xml
---
.github/workflows/daily-it.yml | 359 ---------------------
.github/workflows/pipe-it.yml | 350 --------------------
integration-test/pom.xml | 4 +
.../it/env/cluster/config/MppCommonConfig.java | 6 -
.../env/cluster/config/MppSharedCommonConfig.java | 7 -
.../it/env/remote/config/RemoteCommonConfig.java | 5 -
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 -
.../IoTDBSubscriptionTopicOwnerPartitionIT.java | 1 -
...ConsensusSubscriptionColumnFilterClusterIT.java | 1 -
.../it/local/AbstractSubscriptionLocalIT.java | 2 -
.../it/local/IoTDBSubscriptionTopicOwnerIT.java | 1 -
.../apache/iotdb/db/conf/DataNodeMemoryConfig.java | 5 +-
.../agent/SubscriptionReceiverAgent.java | 16 +-
.../iotdb/db/conf/DataNodeMemoryConfigTest.java | 57 +++-
...rnalRPCServiceImplSubscriptionDisabledTest.java | 34 +-
.../agent/SubscriptionReceiverAgentTest.java | 39 +--
.../ConsensusSubscriptionSetupHandlerTest.java | 13 +-
.../apache/iotdb/commons/conf/CommonConfig.java | 9 +-
.../iotdb/commons/conf/CommonDescriptor.java | 4 -
.../iotdb/commons/conf/CommonConfigTest.java | 16 +
pom.xml | 4 +
21 files changed, 132 insertions(+), 803 deletions(-)
diff --git a/.github/workflows/daily-it.yml b/.github/workflows/daily-it.yml
index 745ffbd0769..553cf6f7514 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 2a9fbbd1239..7b5d6df0b9e 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 4dfea252478..2152a5e1134 100644
--- a/integration-test/pom.xml
+++ b/integration-test/pom.xml
@@ -324,6 +324,10 @@
<configuration>
<groups>${integrationTest.includedGroups}</groups>
<excludedGroups>${integrationTest.excludedGroups}</excludedGroups>
+ <!-- Subscription IT is maintained and executed by
the enterprise edition. -->
+ <excludes>
+ <exclude>**/subscription/it/**</exclude>
+ </excludes>
<useSystemClassLoader>false</useSystemClassLoader>
<parallel>none</parallel>
<forkCount>${integrationTest.forkCount}</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 ec67e5f451b..31f26def318 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 class MppCommonConfig extends MppBaseConfig
implements CommonConfig {
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 36d06aefe88..da61b00e8b1 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 class MppSharedCommonConfig implements CommonConfig
{
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 752dcd009db..1eb7757da4f 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 class RemoteCommonConfig implements CommonConfig {
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 0ad3c23af16..f5324f36101 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 @@ public interface CommonConfig {
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/IoTDBSubscriptionTopicOwnerPartitionIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionTopicOwnerPartitionIT.java
index 31fa3fde022..69ddc56297c 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 class IoTDBSubscriptionTopicOwnerPartitionIT extends
AbstractSubscription
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/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java
index 6c33532828b..e90d94af7ea 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 class IoTDBConsensusSubscriptionColumnFilterClusterIT
extends AbstractSub
.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/local/AbstractSubscriptionLocalIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java
index fe667480c86..c177d039327 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 c08f151c16f..9b506c04951 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 class IoTDBSubscriptionTopicOwnerIT extends
AbstractSubscriptionIT {
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/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 c63cd2c7ee8..a6fe6e0ef99 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 @@ public class DataNodeMemoryConfig {
long maxMemoryAvailable =
queryEngineMemoryManager.getTotalMemorySizeInBytes();
String queryMemoryAllocateProportion =
properties.getProperty("chunk_timeseriesmeta_free_memory_proportion");
- boolean subscriptionEnabled =
- Boolean.parseBoolean(
- properties.getProperty("subscription_enabled",
Boolean.TRUE.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 1f26397cbcc..3fdadb6e84e 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.IoTDBThreadPoolFactory;
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.Objects;
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<ConsumerIdentity, SubscriptionReceiver>
consumerReceivers =
new ConcurrentHashMap<>();
+ private final BooleanSupplier subscriptionEnabledSupplier;
private final ScheduledExecutorService receiverTimeoutChecker;
SubscriptionReceiverAgent() {
- this(SubscriptionReceiverV1::new,
SubscriptionConfig.getInstance().getSubscriptionEnabled());
+ this(
+ SubscriptionReceiverV1::new,
+ SubscriptionConfig.getInstance().getSubscriptionEnabled(),
+ () -> SubscriptionConfig.getInstance().getSubscriptionEnabled());
}
SubscriptionReceiverAgent(
final Supplier<SubscriptionReceiver> 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 class SubscriptionReceiverAgent {
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 6c3e1b83712..6aeeb22f2f6 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 @@ package org.apache.iotdb.db.conf;
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 @@ import static org.junit.Assert.assertTrue;
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 class DataNodeMemoryConfigTest {
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 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(0,
memoryConfig.getSubscriptionMemoryManager().getTotalMemorySizeInBytes());
+ assertEquals(1_000_000L,
memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes());
}
@Test
@@ -133,4 +177,15 @@ public class DataNodeMemoryConfigTest {
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/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java
index 4b61cd72bfd..ecad06aafcc 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.protocol.thrift.impl;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.service.DataNode.DataNodeContext;
import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressResp;
@@ -39,26 +38,19 @@ public class
DataNodeInternalRPCServiceImplSubscriptionDisabledTest {
@Test
public void testSubscriptionRuntimeRPCsAreNoOpWhenSubscriptionIsDisabled() {
- final boolean subscriptionEnabled =
- CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
- try {
- CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false);
- final DataNodeInternalRPCServiceImpl service =
- new
DataNodeInternalRPCServiceImpl(Mockito.mock(DataNodeContext.class));
+ final DataNodeInternalRPCServiceImpl service =
+ new
DataNodeInternalRPCServiceImpl(Mockito.mock(DataNodeContext.class));
- final TPullCommitProgressResp pullResp =
service.pullCommitProgress(null);
- Assert.assertEquals(
- TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(),
pullResp.getStatus().getCode());
- Assert.assertTrue(pullResp.isSetCommitRegionProgress());
- Assert.assertTrue(pullResp.getCommitRegionProgress().isEmpty());
- Assert.assertEquals(
- TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(),
- service.syncSubscriptionProgress(null).getCode());
- Assert.assertEquals(
- TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(),
- service.pushSubscriptionRuntime(null).getCode());
- } finally {
-
CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
- }
+ final TPullCommitProgressResp pullResp = service.pullCommitProgress(null);
+ Assert.assertEquals(
+ TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(),
pullResp.getStatus().getCode());
+ Assert.assertTrue(pullResp.isSetCommitRegionProgress());
+ Assert.assertTrue(pullResp.getCommitRegionProgress().isEmpty());
+ Assert.assertEquals(
+ TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(),
+ service.syncSubscriptionProgress(null).getCode());
+ Assert.assertEquals(
+ TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(),
+ service.pushSubscriptionRuntime(null).getCode());
}
}
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 29d4f4f47f8..bdf75a3a339 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;
@@ -53,29 +52,34 @@ import java.util.function.Supplier;
public class SubscriptionReceiverAgentTest {
@Test
- public void testTimeoutCheckerIsNotScheduledWhenSubscriptionIsDisabled()
throws Exception {
- final boolean subscriptionEnabled =
- CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
- try {
- CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false);
+ public void testDisabledSubscriptionRejectsRequest() throws IOException {
+ final SubscriptionReceiverAgent agent =
+ new SubscriptionReceiverAgent(
+ () -> {
+ throw new AssertionError(
+ "Receiver must not be created when subscription is
disabled");
+ },
+ false,
+ () -> false);
- final SubscriptionReceiverAgent agent = new SubscriptionReceiverAgent();
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR.getStatusCode(),
+ agent.handle(createHandshakeRequest("group", "consumer"),
"root").getStatus().getCode());
+ }
- Assert.assertNull(getReceiverTimeoutChecker(agent));
- } finally {
-
CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
- }
+ @Test
+ public void testTimeoutCheckerIsNotScheduledWhenSubscriptionIsDisabled()
throws Exception {
+ final SubscriptionReceiverAgent agent = new SubscriptionReceiverAgent();
+
+ Assert.assertNull(getReceiverTimeoutChecker(agent));
}
@Test
public void testTimeoutCheckerIsScheduledWhenSubscriptionIsEnabled() throws
Exception {
- final boolean subscriptionEnabled =
- CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
SubscriptionReceiverAgent agent = null;
try {
- CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(true);
-
- agent = new SubscriptionReceiverAgent();
+ agent =
+ new SubscriptionReceiverAgent(() -> new
FakeSubscriptionReceiver(true), true, () -> true);
Assert.assertNotNull(getReceiverTimeoutChecker(agent));
} finally {
@@ -85,7 +89,6 @@ public class SubscriptionReceiverAgentTest {
receiverTimeoutChecker.shutdownNow();
}
}
-
CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
}
}
@@ -213,7 +216,7 @@ public class SubscriptionReceiverAgentTest {
receivers.add(receiver);
return receiver;
};
- return new SubscriptionReceiverAgent(constructor, false);
+ return new SubscriptionReceiverAgent(constructor, false, () -> true);
}
private ScheduledExecutorService getReceiverTimeoutChecker(final
SubscriptionReceiverAgent agent)
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
index 3e0a0351ff8..30cbd487668 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.subscription.broker.consensus;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -52,16 +51,8 @@ public class ConsensusSubscriptionSetupHandlerTest {
@Test
public void testRuntimeUpdatesAreIgnoredWhenSubscriptionIsDisabled() {
- final boolean subscriptionEnabled =
- CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled();
- try {
- CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false);
-
- ConsensusSubscriptionSetupHandler.applyRuntimeState(null, null);
- ConsensusSubscriptionSetupHandler.onRegionRouteChanged(null, 0);
- } finally {
-
CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled);
- }
+ ConsensusSubscriptionSetupHandler.applyRuntimeState(null, null);
+ ConsensusSubscriptionSetupHandler.onRegionRouteChanged(null, 0);
}
@Test
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 2b756199413..cb5902986b8 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 = true;
-
private float subscriptionCacheMemoryUsagePercentage = 0.2F;
private int subscriptionSubtaskExecutorMaxThreadNum = 2;
private int subscriptionConsensusPrefetchExecutorMaxThreadNum = 2;
@@ -2545,11 +2544,7 @@ public class CommonConfig {
}
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 7de383a0d61..b249dea07ee 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 class CommonDescriptor {
}
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 ac3a001175a..2a38ad388d5 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
@@ -24,8 +24,24 @@ import org.apache.iotdb.commons.cluster.NodeStatus;
import org.junit.Assert;
import org.junit.Test;
+import java.io.IOException;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+
public class CommonConfigTest {
+ @Test
+ public void testSubscriptionDisabledInCommonConfig() {
+ assertFalse(CommonConfig.SUBSCRIPTION_ENABLED);
+ assertFalse(new CommonConfig().getSubscriptionEnabled());
+ }
+
+ @Test
+ public void testSubscriptionIsNotExposedInConfigurationTemplate() throws
IOException {
+
assertNull(ConfigurationFileUtils.getConfigurationDefaultValue("subscription_enabled"));
+ }
+
@Test
public void testSameNodeStatusDoesNotClearStatusReason() {
CommonConfig config = new CommonConfig();
diff --git a/pom.xml b/pom.xml
index e427d8680a7..233574ab8f2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -717,6 +717,10 @@
<version>3.1.2</version>
<configuration>
<argLine>${argLine} -Xmx1024m</argLine>
+ <!-- Subscription UT is maintained and executed by the
enterprise edition. -->
+ <excludes>
+ <exclude>**/subscription/**</exclude>
+ </excludes>
<!-- Force the working directory to be different for
each fork, so forks don't trample each other -->
<workingDirectory>${project.build.directory}/fork_${surefire.forkNumber}</workingDirectory>
</configuration>
