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/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/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/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/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..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.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 1f26397cbcc58..3fdadb6e84e85 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, SubscriptionConfig.getInstance().getSubscriptionEnabled()); + this( + SubscriptionReceiverV1::new, + SubscriptionConfig.getInstance().getSubscriptionEnabled(), + () -> 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 6c3e1b837129a..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 @@ -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 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 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/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 4b61cd72bfd28..ecad06aafcc3f 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 static void setUp() { @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 29d4f4f47f818..bdf75a3a33952 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 @@ 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 void testTimeoutCheckerIsScheduledWhenSubscriptionIsEnabled() throws Exce receiverTimeoutChecker.shutdownNow(); } } - CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled); } } @@ -213,7 +216,7 @@ private SubscriptionReceiverAgent createAgent( 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 3e0a0351ff842..30cbd487668a7 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 2b756199413e2..cb5902986b8e8 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 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 ac3a001175afa..2a38ad388d50e 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.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 e427d8680a76e..233574ab8f2c5 100644 --- a/pom.xml +++ b/pom.xml @@ -717,6 +717,10 @@ 3.1.2 ${argLine} -Xmx1024m + + + **/subscription/** + ${project.build.directory}/fork_${surefire.forkNumber}