diff --git a/.github/workflows/daily-it.yml b/.github/workflows/daily-it.yml index 745ffbd0769fe..025a15bbdef76 100644 --- a/.github/workflows/daily-it.yml +++ b/.github/workflows/daily-it.yml @@ -85,995 +85,3 @@ jobs: name: table-standalone-log-java${{ matrix.java }}-${{ runner.os }} path: integration-test/target/cluster-logs retention-days: 3 - PipeSingle: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - cluster2: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - ] - os: [ubuntu-latest] - exclude: - - cluster1: LightWeightStandaloneMode - cluster2: LightWeightStandaloneMode - - cluster1: LightWeightStandaloneMode - cluster2: ScalableSingleNodeMode - - cluster1: ScalableSingleNodeMode - cluster2: LightWeightStandaloneMode - - cluster1: ScalableSingleNodeMode - cluster2: HighPerformanceMode - - cluster1: HighPerformanceMode - cluster2: LightWeightStandaloneMode - - cluster1: HighPerformanceMode - cluster2: HighPerformanceMode - - cluster1: IoTConsensusV2BatchMode - cluster2: LightWeightStandaloneMode - - cluster1: IoTConsensusV2BatchMode - cluster2: HighPerformanceMode - - cluster1: IoTConsensusV2StreamMode - cluster2: LightWeightStandaloneMode - - cluster1: IoTConsensusV2StreamMode - cluster2: HighPerformanceMode - 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 -PMultiClusterIT1 \ - -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-single-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - PipeDualTreeAutoBasic: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - 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.cluster }},${{ matrix.cluster }} \ - -pl integration-test \ - -am -PMultiClusterIT2DualTreeAutoBasic \ - -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-dual-tree-auto-basic-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }} - path: integration-test/target/cluster-logs - retention-days: 30 - PipeDualTreeAutoEnhanced: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - cluster2: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - ] - os: [ubuntu-latest] - exclude: - - cluster1: LightWeightStandaloneMode - cluster2: LightWeightStandaloneMode - - cluster1: LightWeightStandaloneMode - cluster2: ScalableSingleNodeMode - - cluster1: ScalableSingleNodeMode - cluster2: LightWeightStandaloneMode - - cluster1: ScalableSingleNodeMode - cluster2: HighPerformanceMode - - cluster1: HighPerformanceMode - cluster2: LightWeightStandaloneMode - - cluster1: HighPerformanceMode - cluster2: HighPerformanceMode - - cluster1: IoTConsensusV2BatchMode - cluster2: LightWeightStandaloneMode - - cluster1: IoTConsensusV2BatchMode - cluster2: HighPerformanceMode - - cluster1: IoTConsensusV2StreamMode - cluster2: LightWeightStandaloneMode - - cluster1: IoTConsensusV2StreamMode - cluster2: HighPerformanceMode - 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 -PMultiClusterIT2DualTreeAutoEnhanced \ - -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-dual-tree-auto-enhanced-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - PipeDualTreeManual: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - cluster2: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - ] - os: [ubuntu-latest] - exclude: - - cluster1: LightWeightStandaloneMode - cluster2: LightWeightStandaloneMode - - cluster1: LightWeightStandaloneMode - cluster2: ScalableSingleNodeMode - - cluster1: ScalableSingleNodeMode - cluster2: LightWeightStandaloneMode - - cluster1: ScalableSingleNodeMode - cluster2: HighPerformanceMode - - cluster1: HighPerformanceMode - cluster2: LightWeightStandaloneMode - - cluster1: HighPerformanceMode - cluster2: HighPerformanceMode - - cluster1: IoTConsensusV2BatchMode - cluster2: LightWeightStandaloneMode - - cluster1: IoTConsensusV2BatchMode - cluster2: HighPerformanceMode - - cluster1: IoTConsensusV2StreamMode - cluster2: LightWeightStandaloneMode - - cluster1: IoTConsensusV2StreamMode - cluster2: HighPerformanceMode - 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 -PMultiClusterIT2DualTreeManual \ - -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-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 - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - 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.cluster }},${{ matrix.cluster }} \ - -pl integration-test \ - -am -PMultiClusterIT2DualTableManualBasic \ - -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-dual-table-manual-basic-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }} - path: integration-test/target/cluster-logs - retention-days: 30 - PipeDualTableManualEnhanced: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster: - [ - LightWeightStandaloneMode, - ScalableSingleNodeMode, - HighPerformanceMode, - IoTConsensusV2BatchMode, - IoTConsensusV2StreamMode, - ] - 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.cluster }},${{ matrix.cluster }} \ - -pl integration-test \ - -am -PMultiClusterIT2DualTableManualEnhanced \ - -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-dual-table-manual-enhanced-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }} - path: integration-test/target/cluster-logs - retention-days: 30 diff --git a/.github/workflows/pipe-it.yml b/.github/workflows/pipe-it.yml deleted file mode 100644 index 2a9fbbd12398b..0000000000000 --- a/.github/workflows/pipe-it.yml +++ /dev/null @@ -1,1102 +0,0 @@ -name: Multi-Cluster IT - -on: - push: - branches: - - master - - "rel/*" - - "rc/*" - paths-ignore: - - "docs/**" - - "site/**" - - "iotdb-client/client-cpp/**" - - ".github/workflows/client-cpp-package.yml" - - ".github/scripts/package-client-cpp-*.sh" - - ".github/workflows/multi-language-client.yml" - - "iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/**" #queryengine - pull_request: - branches: - - master - - "rel/*" - - "rc/*" - - "force_ci/**" - paths-ignore: - - "docs/**" - - "site/**" - - "iotdb-client/client-cpp/**" - - ".github/workflows/client-cpp-package.yml" - - ".github/scripts/package-client-cpp-*.sh" - - ".github/workflows/multi-language-client.yml" - - "iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/**" #queryengine - # allow manually run the action: - workflow_dispatch: - -concurrency: - group: ${{ github.workflow }}-${{ github.ref }} - cancel-in-progress: true - -env: - MAVEN_OPTS: -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 - MAVEN_ARGS: --batch-mode --no-transfer-progress - DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }} - -jobs: - single: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: [HighPerformanceMode] - cluster2: [HighPerformanceMode] - 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 -PMultiClusterIT1 \ - -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-single-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - # 12 IT classes split across 3 parallel shards to cut the historical ~42 min - # wall clock to ~14 min. See cluster-it-1c1d.yml for the shard pattern. - dual-tree-auto-basic: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster: [HighPerformanceMode] - os: [ubuntu-latest] - shard: [0, 1, 2] - 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: Build IT shard list - shell: bash - # See cluster-it-1c1d.yml for the shard-list pattern. Write under - # $RUNNER_TEMP (outside the repo) so Apache RAT doesn't flag the file. - run: | - set -euo pipefail - SHARD=${{ matrix.shard }} - TOTAL=3 - grep -rlE --include='*IT.java' '\bMultiClusterIT2DualTreeAutoBasic\b' integration-test/src/test/java \ - | awk -F'/' '{print $NF}' | sed 's/\.java$//' \ - | sort \ - | awk -v s=$SHARD -v t=$TOTAL 'NR%t==s' \ - > "$RUNNER_TEMP/it-shard.txt" - echo "Shard $SHARD/$TOTAL contains $(wc -l < "$RUNNER_TEMP/it-shard.txt") test classes" - head -5 "$RUNNER_TEMP/it-shard.txt" - - 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.cluster }},${{ matrix.cluster }} \ - -Dfailsafe.includesFile="$RUNNER_TEMP/it-shard.txt" \ - -DfailIfNoTests=false \ - -Dfailsafe.failIfNoSpecifiedTests=false \ - -pl integration-test \ - -am -PMultiClusterIT2DualTreeAutoBasic \ - -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-dual-tree-auto-basic-shard${{ matrix.shard }}-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }} - path: integration-test/target/cluster-logs - retention-days: 30 - # 9 IT classes split across 3 parallel shards to cut the historical ~51 min - # wall clock to ~17 min. See cluster-it-1c1d.yml for the shard pattern. - dual-tree-auto-enhanced: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: [HighPerformanceMode] - cluster2: [HighPerformanceMode] - os: [ubuntu-latest] - shard: [0, 1, 2] - 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: Build IT shard list - shell: bash - # See cluster-it-1c1d.yml for the shard-list pattern. Write under - # $RUNNER_TEMP (outside the repo) so Apache RAT doesn't flag the file. - run: | - set -euo pipefail - SHARD=${{ matrix.shard }} - TOTAL=3 - grep -rlE --include='*IT.java' '\bMultiClusterIT2DualTreeAutoEnhanced\b' integration-test/src/test/java \ - | awk -F'/' '{print $NF}' | sed 's/\.java$//' \ - | sort \ - | awk -v s=$SHARD -v t=$TOTAL 'NR%t==s' \ - > "$RUNNER_TEMP/it-shard.txt" - echo "Shard $SHARD/$TOTAL contains $(wc -l < "$RUNNER_TEMP/it-shard.txt") test classes" - head -5 "$RUNNER_TEMP/it-shard.txt" - - 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 }} \ - -Dfailsafe.includesFile="$RUNNER_TEMP/it-shard.txt" \ - -DfailIfNoTests=false \ - -Dfailsafe.failIfNoSpecifiedTests=false \ - -pl integration-test \ - -am -PMultiClusterIT2DualTreeAutoEnhanced \ - -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-dual-tree-auto-enhanced-shard${{ matrix.shard }}-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }} - path: integration-test/target/cluster-logs - retention-days: 30 - # 11 IT classes split across 3 parallel shards to cut the historical ~27 min - # wall clock to ~9 min. See cluster-it-1c1d.yml for the shard pattern. - dual-tree-manual: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster1: [HighPerformanceMode] - cluster2: [HighPerformanceMode] - os: [ubuntu-latest] - shard: [0, 1, 2] - 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: Build IT shard list - shell: bash - # See cluster-it-1c1d.yml for the shard-list pattern. Write under - # $RUNNER_TEMP (outside the repo) so Apache RAT doesn't flag the file. - run: | - set -euo pipefail - SHARD=${{ matrix.shard }} - TOTAL=3 - grep -rlE --include='*IT.java' '\bMultiClusterIT2DualTreeManual\b' integration-test/src/test/java \ - | awk -F'/' '{print $NF}' | sed 's/\.java$//' \ - | sort \ - | awk -v s=$SHARD -v t=$TOTAL 'NR%t==s' \ - > "$RUNNER_TEMP/it-shard.txt" - echo "Shard $SHARD/$TOTAL contains $(wc -l < "$RUNNER_TEMP/it-shard.txt") test classes" - head -5 "$RUNNER_TEMP/it-shard.txt" - - 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 }} \ - -Dfailsafe.includesFile="$RUNNER_TEMP/it-shard.txt" \ - -DfailIfNoTests=false \ - -Dfailsafe.failIfNoSpecifiedTests=false \ - -pl integration-test \ - -am -PMultiClusterIT2DualTreeManual \ - -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-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 - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster: [HighPerformanceMode] - os: [ubuntu-latest] - shard: [0, 1, 2] - 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: Build IT shard list - shell: bash - # See cluster-it-1c1d.yml for the shard-list pattern. Write under - # $RUNNER_TEMP (outside the repo) so Apache RAT doesn't flag the file. - run: | - set -euo pipefail - SHARD=${{ matrix.shard }} - TOTAL=3 - grep -rlE --include='*IT.java' '\bMultiClusterIT2DualTableManualBasic\b' integration-test/src/test/java \ - | awk -F'/' '{print $NF}' | sed 's/\.java$//' \ - | sort \ - | awk -v s=$SHARD -v t=$TOTAL 'NR%t==s' \ - > "$RUNNER_TEMP/it-shard.txt" - echo "Shard $SHARD/$TOTAL contains $(wc -l < "$RUNNER_TEMP/it-shard.txt") test classes" - head -5 "$RUNNER_TEMP/it-shard.txt" - - 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.cluster }},${{ matrix.cluster }} \ - -Dfailsafe.includesFile="$RUNNER_TEMP/it-shard.txt" \ - -DfailIfNoTests=false \ - -Dfailsafe.failIfNoSpecifiedTests=false \ - -pl integration-test \ - -am -PMultiClusterIT2DualTableManualBasic \ - -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-dual-table-manual-basic-shard${{ matrix.shard }}-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }} - path: integration-test/target/cluster-logs - retention-days: 30 - # 11 IT classes split across 3 parallel shards to cut the historical ~62 min - # wall clock to ~22 min. See cluster-it-1c1d.yml for the shard pattern. - dual-table-manual-enhanced: - strategy: - fail-fast: false - max-parallel: 15 - matrix: - java: [17] - # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet. - cluster: [HighPerformanceMode] - os: [ubuntu-latest] - shard: [0, 1, 2] - 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: Build IT shard list - shell: bash - # See cluster-it-1c1d.yml for the shard-list pattern. Write under - # $RUNNER_TEMP (outside the repo) so Apache RAT doesn't flag the file. - run: | - set -euo pipefail - SHARD=${{ matrix.shard }} - TOTAL=3 - grep -rlE --include='*IT.java' '\bMultiClusterIT2DualTableManualEnhanced\b' integration-test/src/test/java \ - | awk -F'/' '{print $NF}' | sed 's/\.java$//' \ - | sort \ - | awk -v s=$SHARD -v t=$TOTAL 'NR%t==s' \ - > "$RUNNER_TEMP/it-shard.txt" - echo "Shard $SHARD/$TOTAL contains $(wc -l < "$RUNNER_TEMP/it-shard.txt") test classes" - head -5 "$RUNNER_TEMP/it-shard.txt" - - 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.cluster }},${{ matrix.cluster }} \ - -Dfailsafe.includesFile="$RUNNER_TEMP/it-shard.txt" \ - -DfailIfNoTests=false \ - -Dfailsafe.failIfNoSpecifiedTests=false \ - -pl integration-test \ - -am -PMultiClusterIT2DualTableManualEnhanced \ - -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-dual-table-manual-enhanced-shard${{ matrix.shard }}-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }} - path: integration-test/target/cluster-logs - retention-days: 30 - triple: - strategy: - fail-fast: false - max-parallel: 1 - matrix: - java: [ 17 ] - cluster1: [ ScalableSingleNodeMode ] - cluster2: [ ScalableSingleNodeMode ] - cluster3: [ 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 }},${{ matrix.cluster3 }} \ - -pl integration-test \ - -am -PMultiClusterIT3 \ - -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-triple-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}-${{ matrix.cluster3 }} - path: integration-test/target/cluster-logs - retention-days: 30 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/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java index 9a6fe6ddf41d1..ea9a61f250f5b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java @@ -358,11 +358,6 @@ public void testAsyncConnectorUseNodeUrls() throws Exception { doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName()); } - @Test - public void testAirGapConnectorUseNodeUrls() throws Exception { - doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName()); - } - private void doTestUseNodeUrls(String connectorName) throws Exception { senderEnv .getConfig() @@ -407,16 +402,7 @@ private void doTestUseNodeUrls(String connectorName) throws Exception { }; for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) { - if (connectorName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) { - // Use default port for convenience - nodeUrlsBuilder - .append(wrapper.getIp()) - .append(":") - .append(wrapper.getPipeAirGapReceiverPort()) - .append(","); - } else { - nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(","); - } + nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(","); } try (final SyncConfigNodeIServiceClient client = diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java index 5489c0691b360..ed04af9f95aa3 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java @@ -353,11 +353,6 @@ public void testAsyncConnectorUseNodeUrls() throws Exception { doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName()); } - @Test - public void testAirGapConnectorUseNodeUrls() throws Exception { - doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName()); - } - private void doTestUseNodeUrls(String sinkName) throws Exception { senderEnv .getConfig() @@ -398,16 +393,7 @@ private void doTestUseNodeUrls(String sinkName) throws Exception { final StringBuilder nodeUrlsBuilder = new StringBuilder(); for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) { - if (sinkName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) { - // Use default port for convenience - nodeUrlsBuilder - .append(wrapper.getIp()) - .append(":") - .append(wrapper.getPipeAirGapReceiverPort()) - .append(","); - } else { - nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(","); - } + nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(","); } try (final SyncConfigNodeIServiceClient client = diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java index 22483ab43a68d..6f92999a71c88 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java @@ -795,8 +795,7 @@ public void testInformationSchema() throws SQLException { "test,view_table,s11,INT32,FIELD,USING,null,", "test,view_table,s3,INT32,FIELD,USING,null,"))); - statement.execute( - "create pipe a2b with source('double-living'='true') with sink ('sink'='write-back-sink')"); + statement.execute("create pipe a2b with sink ('sink'='do-nothing-sink')"); TestUtils.assertResultSetEqual( statement.executeQuery("select id from pipes where creation_time > 0"), "id,", @@ -980,8 +979,7 @@ public void testDBAuth() throws SQLException { final Statement adminStmt = adminCon.createStatement()) { adminStmt.execute("create user test 'password123456'"); adminStmt.execute("create database db"); - adminStmt.execute( - "create pipe a2b with source('double-living'='true') with sink ('sink'='write-back-sink')"); + adminStmt.execute("create pipe a2b with sink ('sink'='do-nothing-sink')"); } try (final Connection userCon = 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/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionSinkConstructor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionSinkConstructor.java index c7222e9a1bc97..82c394c7fb545 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionSinkConstructor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionSinkConstructor.java @@ -22,7 +22,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink; import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor; -import org.apache.iotdb.confignode.manager.pipe.sink.protocol.IoTDBConfigRegionAirGapSink; import org.apache.iotdb.confignode.manager.pipe.sink.protocol.IoTDBConfigRegionSink; import org.apache.iotdb.pipe.api.PipeConnector; @@ -41,9 +40,6 @@ protected void initConstructors() { pluginConstructors.put( BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(), IoTDBConfigRegionSink::new); - pluginConstructors.put( - BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(), - IoTDBConfigRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new); @@ -55,8 +51,6 @@ protected void initConstructors() { BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(), IoTDBConfigRegionSink::new); pluginConstructors.put( BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(), IoTDBConfigRegionSink::new); - pluginConstructors.put( - BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBConfigRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new); } diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 0645db5cf6d39..a7e07cbc1943d 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2573,6 +2573,8 @@ private DataNodePipeMessages() {} public static final String MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD = "Transfer file %s error, result status %s."; + public static final String EXCEPTION_THE_PARAMETER_ARG_CANNOT_BE_SET_TO_FALSE_111C3408 = + "The parameter %s cannot be set to false."; public static final String EXCEPTION_LEGACY_PIPE_RECEIVER_REQUIRES_A_LOGGED_IN_SESSION_D96219BF = "Legacy pipe receiver requires a logged-in session."; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 65c71850652f9..6154ebd4c5db7 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2402,6 +2402,8 @@ private DataNodePipeMessages() {} public static final String MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD = "传输文件 %s 出错，结果状态为 %s。"; + public static final String EXCEPTION_THE_PARAMETER_ARG_CANNOT_BE_SET_TO_FALSE_111C3408 = + "参数 %s 不能设置为 false。"; public static final String EXCEPTION_LEGACY_PIPE_RECEIVER_REQUIRES_A_LOGGED_IN_SESSION_D96219BF = "Legacy pipe receiver 需要已登录的 session。"; } 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 350f5d780197f..b9dc9f410a9dd 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; @@ -521,9 +522,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/pipe/agent/plugin/dataregion/PipeDataRegionProcessorConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionProcessorConstructor.java index a6160e6b90088..b83a75460f898 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionProcessorConstructor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionProcessorConstructor.java @@ -21,18 +21,10 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.throwing.ThrowingExceptionProcessor; import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeProcessorConstructor; import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper; -import org.apache.iotdb.db.pipe.processor.aggregate.AggregateProcessor; -import org.apache.iotdb.db.pipe.processor.aggregate.operator.processor.StandardStatisticsOperatorProcessor; -import org.apache.iotdb.db.pipe.processor.aggregate.window.processor.TumblingWindowingProcessor; -import org.apache.iotdb.db.pipe.processor.downsampling.changing.ChangingValueSamplingProcessor; -import org.apache.iotdb.db.pipe.processor.downsampling.sdt.SwingingDoorTrendingSamplingProcessor; -import org.apache.iotdb.db.pipe.processor.downsampling.tumbling.TumblingTimeSamplingProcessor; import org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor; import org.apache.iotdb.db.pipe.processor.schemachange.RenameDatabaseProcessor; -import org.apache.iotdb.db.pipe.processor.twostage.plugin.TwoStageCountProcessor; class PipeDataRegionProcessorConstructor extends PipeProcessorConstructor { @@ -44,28 +36,6 @@ class PipeDataRegionProcessorConstructor extends PipeProcessorConstructor { protected void initConstructors() { pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName(), DoNothingProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.TUMBLING_TIME_SAMPLING_PROCESSOR.getPipePluginName(), - TumblingTimeSamplingProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.SDT_SAMPLING_PROCESSOR.getPipePluginName(), - SwingingDoorTrendingSamplingProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.CHANGING_VALUE_SAMPLING_PROCESSOR.getPipePluginName(), - ChangingValueSamplingProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.THROWING_EXCEPTION_PROCESSOR.getPipePluginName(), - ThrowingExceptionProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.AGGREGATE_PROCESSOR.getPipePluginName(), AggregateProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.STANDARD_STATISTICS_PROCESSOR.getPipePluginName(), - StandardStatisticsOperatorProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.TUMBLING_WINDOWING_PROCESSOR.getPipePluginName(), - TumblingWindowingProcessor::new); - pluginConstructors.put( - BuiltinPipePlugin.COUNT_POINT_PROCESSOR.getPipePluginName(), TwoStageCountProcessor::new); pluginConstructors.put( BuiltinPipePlugin.IOT_CONSENSUS_V2_PROCESSOR.getPipePluginName(), IoTConsensusV2Processor::new); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java index c8e87890afd69..1838f4958845a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java @@ -23,11 +23,8 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink; import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor; import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper; -import org.apache.iotdb.db.pipe.sink.protocol.airgap.IoTDBDataRegionAirGapSink; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink; -import org.apache.iotdb.db.pipe.sink.protocol.opcda.OpcDaSink; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink; import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketSink; @@ -63,13 +60,8 @@ protected void initConstructors() { pluginConstructors.put( BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName(), IoTDBLegacyPipeSink::new); - pluginConstructors.put( - BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(), - IoTDBDataRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(), WebSocketSink::new); - pluginConstructors.put(BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), OpcUaSink::new); - pluginConstructors.put(BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName(), OpcDaSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new); pluginConstructors.put( @@ -86,12 +78,8 @@ protected void initConstructors() { IoTDBDataRegionAsyncSink::new); pluginConstructors.put( BuiltinPipePlugin.IOTDB_LEGACY_PIPE_SINK.getPipePluginName(), IoTDBLegacyPipeSink::new); - pluginConstructors.put( - BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBDataRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(), WebSocketSink::new); - pluginConstructors.put(BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), OpcUaSink::new); - pluginConstructors.put(BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(), OpcDaSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new); pluginConstructors.put( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionSinkConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionSinkConstructor.java index 160ecf54c0158..a7752994ee264 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionSinkConstructor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionSinkConstructor.java @@ -22,7 +22,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink; import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor; -import org.apache.iotdb.db.pipe.sink.protocol.airgap.IoTDBSchemaRegionAirGapSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBSchemaRegionSink; import org.apache.iotdb.pipe.api.PipeConnector; @@ -41,9 +40,6 @@ protected void initConstructors() { pluginConstructors.put( BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(), IoTDBSchemaRegionSink::new); - pluginConstructors.put( - BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(), - IoTDBSchemaRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new); @@ -55,8 +51,6 @@ protected void initConstructors() { BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(), IoTDBSchemaRegionSink::new); pluginConstructors.put( BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(), IoTDBSchemaRegionSink::new); - pluginConstructors.put( - BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBSchemaRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java index 1d94d4d941902..67cbec279927f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java @@ -78,7 +78,7 @@ public synchronized String register( connectorKey); final boolean isDataRegionSink = isDataRegionSink(environment.getRegionId()); - final int sinkNum = calculateSinkSubtaskNum(pipeSinkParameters, isDataRegionSink, connectorKey); + final int sinkNum = calculateSinkSubtaskNum(pipeSinkParameters, isDataRegionSink); boolean realTimeFirst = false; final String attributeSortedString = generateAttributeSortedString(pipeSinkParameters, environment.getRegionId()); @@ -260,8 +260,7 @@ public synchronized boolean hasRegisteredSubtasks( public static int calculateSinkSubtaskNum( final PipeParameters pipeSinkParameters, final int regionId) { - final String connectorKey = getConnectorKey(pipeSinkParameters); - return calculateSinkSubtaskNum(pipeSinkParameters, isDataRegionSink(regionId), connectorKey); + return calculateSinkSubtaskNum(pipeSinkParameters, isDataRegionSink(regionId)); } public static String generateAttributeSortedString( @@ -285,9 +284,7 @@ private static boolean isDataRegionSink(final int regionId) { } private static int calculateSinkSubtaskNum( - final PipeParameters pipeSinkParameters, - final boolean isDataRegionSink, - final String connectorKey) { + final PipeParameters pipeSinkParameters, final boolean isDataRegionSink) { if (!isDataRegionSink) { // Do not allow parallel tasks for schema region connectors to avoid the potential disorder of // the schema region data transfer. @@ -300,9 +297,7 @@ private static int calculateSinkSubtaskNum( Arrays.asList( PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY, PipeSinkConstant.SINK_IOTDB_PARALLEL_TASKS_KEY), - PipeSinkConstant.SINGLE_THREAD_DEFAULT_SINK.contains(connectorKey) - ? 1 - : PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE); + PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE); } private static String generateAttributeSortedString( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java index 3e9c8ea4f7e9d..4f523cb64a371 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePatternOperations; import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; import org.apache.iotdb.commons.pipe.source.IoTDBSource; @@ -51,6 +52,7 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent; import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; import org.apache.iotdb.pipe.api.exception.PipePasswordCheckException; import org.apache.iotdb.rpc.TSStatusCode; @@ -142,6 +144,18 @@ public class IoTDBDataRegionSource extends IoTDBSource { public void validate(final PipeParameterValidator validator) throws Exception { super.validate(validator); + final boolean forwardingPipeRequests = + validator + .getParameters() + .getBooleanOrDefault( + PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS, + PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE); + if (!forwardingPipeRequests) { + throw new PipeParameterNotValidException( + String.format( + DataNodePipeMessages.EXCEPTION_THE_PARAMETER_ARG_CANNOT_BE_SET_TO_FALSE_111C3408, + PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY)); + } final Pair insertionDeletionListeningOptionPair = DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair( validator.getParameters()); @@ -202,32 +216,6 @@ public void validate(final PipeParameterValidator validator) throws Exception { Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY), EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE)); - // Validate source.realtime.mode - if (validator - .getParameters() - .getBooleanOrDefault( - Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY), - EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE) - || validator - .getParameters() - .hasAnyAttributes( - SOURCE_START_TIME_KEY, - EXTRACTOR_START_TIME_KEY, - SOURCE_END_TIME_KEY, - EXTRACTOR_END_TIME_KEY)) { - validator.validateAttributeValueRange( - validator.getParameters().hasAttribute(EXTRACTOR_REALTIME_MODE_KEY) - ? EXTRACTOR_REALTIME_MODE_KEY - : SOURCE_REALTIME_MODE_KEY, - true, - EXTRACTOR_REALTIME_MODE_FILE_VALUE, - EXTRACTOR_REALTIME_MODE_HYBRID_VALUE, - EXTRACTOR_REALTIME_MODE_LOG_VALUE, - EXTRACTOR_REALTIME_MODE_FORCED_LOG_VALUE, - EXTRACTOR_REALTIME_MODE_STREAM_MODE_VALUE, - EXTRACTOR_REALTIME_MODE_BATCH_MODE_VALUE); - } - checkInvalidParameters(validator); constructHistoricalSource(); @@ -392,6 +380,13 @@ private void constructRealtimeSource(final PipeParameters parameters) { return; } + if (!(pipeName != null + && (pipeName.startsWith(PipeStaticMeta.SUBSCRIPTION_PIPE_PREFIX) + || pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)))) { + realtimeSource = new PipeRealtimeDataRegionTsFileSource(); + return; + } + // Use hybrid mode by default if (!parameters.hasAnyAttributes(EXTRACTOR_MODE_STREAMING_KEY, SOURCE_MODE_STREAMING_KEY) && !parameters.hasAnyAttributes(EXTRACTOR_REALTIME_MODE_KEY, SOURCE_REALTIME_MODE_KEY)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java index 732275824eb46..d60cbd05f7c9c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java @@ -120,7 +120,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor { private final AtomicReference> dataRegionTimePartitionIdBound = new AtomicReference<>(); - protected boolean isForwardingPipeRequests; + protected boolean isForwardingPipeRequests = true; private boolean shouldTransferModFile; // Whether to transfer mods @@ -262,15 +262,7 @@ public void customize( ? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) : TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1; - final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters); - if (isDoubleLiving) { - isForwardingPipeRequests = false; - } else { - isForwardingPipeRequests = - parameters.getBooleanOrDefault( - PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS, - PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE); - } + isForwardingPipeRequests = true; if (parameters.hasAnyAttributes(EXTRACTOR_MODS_KEY, SOURCE_MODS_KEY)) { shouldTransferModFile = 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 4b26a5668374a..f97110f6100c7 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; @@ -56,25 +57,32 @@ 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()); private final ThreadLocal receiverThreadLocal = new ThreadLocal<>(); private final Set activeReceivers = ConcurrentHashMap.newKeySet(); - private final ScheduledExecutorService receiverTimeoutChecker = - IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor( - SubscriptionReceiverAgent.class.getSimpleName() + "-Timeout-Checker"); + private final ScheduledExecutorService receiverTimeoutChecker; SubscriptionReceiverAgent() { RECEIVER_CONSTRUCTORS.put( PipeSubscribeRequestVersion.VERSION_1.getVersion(), SubscriptionReceiverV1::new); - ScheduledExecutorUtil.safelyScheduleWithFixedDelay( - receiverTimeoutChecker, - this::checkReceiverTimeouts, - Math.max(1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), - Math.max(1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), - TimeUnit.MILLISECONDS); + if (SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + receiverTimeoutChecker = + IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor( + SubscriptionReceiverAgent.class.getSimpleName() + "-Timeout-Checker"); + ScheduledExecutorUtil.safelyScheduleWithFixedDelay( + receiverTimeoutChecker, + this::checkReceiverTimeouts, + Math.max( + 1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), + Math.max( + 1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), + TimeUnit.MILLISECONDS); + } else { + receiverTimeoutChecker = null; + } } public TPipeSubscribeResp handle(final TPipeSubscribeReq req) { 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 54f4b849ce771..740d822134698 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 @@ -19,12 +19,15 @@ 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.Assert; import org.junit.Test; +import java.lang.reflect.Method; import java.util.Arrays; import static org.junit.Assert.assertArrayEquals; @@ -34,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( @@ -42,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 @@ -106,4 +151,15 @@ public void testRpcMemoryControlIsActivatedOnlyExplicitly() { Assert.assertTrue(MemoryConfig.getInstance().getAutoResizingBufferMemoryTotalSizeInBytes() > 0); } + + 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/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java index 9545d71d96569..6349f442f398d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java @@ -19,7 +19,6 @@ package org.apache.iotdb.db.pipe.agent.task.subtask.sink; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; @@ -72,17 +71,7 @@ public void testCalculateSinkSubtaskNumForDataRegionSink() { } @Test - public void testCalculateSinkSubtaskNumUsesSingleThreadDefaultSinkAndSchemaRegionLimit() { - final Map singleThreadAttributes = new HashMap<>(); - singleThreadAttributes.put( - PipeSinkConstant.CONNECTOR_SERIALIZE_BY_REGION_KEY, Boolean.FALSE.toString()); - singleThreadAttributes.put( - PipeSinkConstant.CONNECTOR_KEY, BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName()); - Assert.assertEquals( - 1, - PipeSinkSubtaskManager.calculateSinkSubtaskNum( - new PipeParameters(singleThreadAttributes), -1)); - + public void testCalculateSinkSubtaskNumUsesSchemaRegionLimit() { final Map schemaRegionAttributes = new HashMap<>(); schemaRegionAttributes.put( PipeSinkConstant.CONNECTOR_SERIALIZE_BY_REGION_KEY, Boolean.FALSE.toString()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java index 34bfc83763b89..214cb40401631 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java @@ -27,7 +27,6 @@ import org.apache.iotdb.db.pipe.event.common.statement.PipeStatementInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink; import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketConnectorServer; @@ -47,7 +46,6 @@ import org.mockito.Mockito; import java.lang.reflect.Field; -import java.security.SecureRandom; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -386,97 +384,6 @@ public void testWebSocketSinkDropDoesNotRequeueDroppedPipeEvents() { } } - @Test - public void testOpcUaSink() { - final List schemaList = - Arrays.asList( - new MeasurementSchema("s1", TSDataType.INT64), - new MeasurementSchema("s2", TSDataType.INT64)); - - final Tablet tablet = new Tablet("root.db.d1.vector6", schemaList, 100); - - long timestamp = System.currentTimeMillis(); - for (long row = 0; row < 100; row++) { - final int rowSize = tablet.getRowSize(); - tablet.addTimestamp(rowSize, timestamp); - for (int i = 0; i < 2; i++) { - tablet.addValue( - schemaList.get(i).getMeasurementName(), rowSize, new SecureRandom().nextLong()); - } - timestamp++; - } - - final List opcSchemaList = - Arrays.asList( - new MeasurementSchema("value1", TSDataType.INT64), - new MeasurementSchema("quality1", TSDataType.BOOLEAN)); - final Tablet qualityTablet = new Tablet("root.db.d1.vector6.s3", opcSchemaList, 100); - - timestamp = System.currentTimeMillis(); - for (long row = 0; row < 100; row++) { - final int rowSize = qualityTablet.getRowSize(); - qualityTablet.addTimestamp(rowSize, timestamp); - qualityTablet.addValue( - opcSchemaList.get(0).getMeasurementName(), rowSize, new SecureRandom().nextLong()); - qualityTablet.addValue(opcSchemaList.get(1).getMeasurementName(), rowSize, true); - timestamp++; - } - - try (final OpcUaSink qualityOPC = new OpcUaSink(); - final OpcUaSink normalOPC = new OpcUaSink()) { - final PipeTaskRuntimeConfiguration configuration = - new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("temp", 0, 1)); - qualityOPC.customize( - new PipeParameters( - new HashMap() { - { - put( - PipeSinkConstant.CONNECTOR_KEY, - BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName()); - put(PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, "true"); - put(PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY, "value1"); - put(PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY, "quality1"); - } - }), - configuration); - normalOPC.customize( - new PipeParameters( - new HashMap() { - { - put( - PipeSinkConstant.CONNECTOR_KEY, - BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName()); - } - }), - configuration); - final PipeRawTabletInsertionEvent event = - new PipeRawTabletInsertionEvent( - false, "root.db", "db", "root.db", tablet, false, "pipe", 0L, null, null, false); - event.increaseReferenceCount(""); - normalOPC.transfer(event); - // Shall not throw - qualityOPC.transfer(event); - event.decreaseReferenceCount("", false); - - qualityOPC.transfer( - new PipeRawTabletInsertionEvent( - false, - "root.db", - "db", - "root.db", - qualityTablet, - false, - "pipe", - 0L, - null, - null, - false)); - - } catch (Exception e) { - Assert.fail(); - } - } - @Test public void testWriteBackSinkTargetDatabaseValidation() throws Exception { assertWriteBackSinkTargetDatabaseValid("target"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java index 1e6f3f865c103..37e8fb5418fc6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java @@ -19,7 +19,6 @@ package org.apache.iotdb.db.pipe.sink.protocol.airgap; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment; @@ -115,9 +114,7 @@ public void testTransferTsFileBatchOverAirGap() throws Exception { private PipeParameters buildParameters(final boolean useTsFileBatch) { final Map attributes = new HashMap<>(); - attributes.put( - PipeSinkConstant.CONNECTOR_KEY, - BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName()); + attributes.put(PipeSinkConstant.CONNECTOR_KEY, "iotdb-air-gap-connector"); attributes.put(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY, "127.0.0.1:6668"); attributes.put(PipeSinkConstant.CONNECTOR_IOTDB_BATCH_DELAY_MS_KEY, "200"); attributes.put(PipeSinkConstant.CONNECTOR_IOTDB_BATCH_SIZE_KEY, "1048576"); 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 91393bf9fd2b4..a3ff82e3b92fc 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 = @@ -374,8 +375,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; @@ -2496,11 +2495,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/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java index 76b45d1e79895..eb23ad70133c9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java @@ -19,27 +19,16 @@ package org.apache.iotdb.commons.pipe.agent.plugin.builtin; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.aggregate.AggregateProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.aggregate.StandardStatisticsProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.aggregate.TumblingWindowingProcessor; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.downsampling.ChangingValueSamplingProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.downsampling.SwingingDoorTrendingSamplingProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.downsampling.TumblingTimeSamplingProcessor; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.iotconsensusv2.IoTConsensusV2Processor; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.schemachange.RenameDatabaseProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.throwing.ThrowingExceptionProcessor; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.twostage.TwoStageCountProcessor; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.airgap.IoTDBAirGapSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.consensus.IoTConsensusV2AsyncSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBLegacyPipeSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftAsyncSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSslSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSyncSink; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcda.OpcDaSink; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcua.OpcUaSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.websocket.WebSocketSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.writeback.WriteBackSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.source.donothing.DoNothingSource; @@ -61,18 +50,8 @@ public enum BuiltinPipePlugin { // processors DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class), - TUMBLING_TIME_SAMPLING_PROCESSOR( - "tumbling-time-sampling-processor", TumblingTimeSamplingProcessor.class), - SDT_SAMPLING_PROCESSOR("sdt-sampling-processor", SwingingDoorTrendingSamplingProcessor.class), - CHANGING_VALUE_SAMPLING_PROCESSOR( - "changing-value-sampling-processor", ChangingValueSamplingProcessor.class), - THROWING_EXCEPTION_PROCESSOR("throwing-exception-processor", ThrowingExceptionProcessor.class), - AGGREGATE_PROCESSOR("aggregate-processor", AggregateProcessor.class), - COUNT_POINT_PROCESSOR("count-point-processor", TwoStageCountProcessor.class), // Hidden-processors, which are plugins of the processors - STANDARD_STATISTICS_PROCESSOR("standard-statistics-processor", StandardStatisticsProcessor.class), - TUMBLING_WINDOWING_PROCESSOR("tumbling-windowing-processor", TumblingWindowingProcessor.class), IOT_CONSENSUS_V2_PROCESSOR("iot-consensus-v2-processor", IoTConsensusV2Processor.class), // Legacy alias for stale PipeMeta written before the PipeConsensus -> IoTConsensusV2 rename. PIPE_CONSENSUS_PROCESSOR("pipe-consensus-processor", IoTConsensusV2Processor.class), @@ -85,15 +64,12 @@ public enum BuiltinPipePlugin { IOTDB_THRIFT_SYNC_CONNECTOR("iotdb-thrift-sync-connector", IoTDBThriftSyncSink.class), IOTDB_THRIFT_ASYNC_CONNECTOR("iotdb-thrift-async-connector", IoTDBThriftAsyncSink.class), IOTDB_LEGACY_PIPE_CONNECTOR("iotdb-legacy-pipe-connector", IoTDBLegacyPipeSink.class), - IOTDB_AIR_GAP_CONNECTOR("iotdb-air-gap-connector", IoTDBAirGapSink.class), IOT_CONSENSUS_V2_ASYNC_CONNECTOR( "iot-consensus-v2-async-connector", IoTConsensusV2AsyncSink.class), // Legacy alias for stale PipeMeta written before the PipeConsensus -> IoTConsensusV2 rename. PIPE_CONSENSUS_ASYNC_CONNECTOR("pipe-consensus-async-connector", IoTConsensusV2AsyncSink.class), WEBSOCKET_CONNECTOR("websocket-connector", WebSocketSink.class), - OPC_UA_CONNECTOR("opc-ua-connector", OpcUaSink.class), - OPC_DA_CONNECTOR("opc-da-connector", OpcDaSink.class), WRITE_BACK_CONNECTOR("write-back-connector", WriteBackSink.class), DO_NOTHING_SINK("do-nothing-sink", DoNothingSink.class), @@ -102,10 +78,7 @@ public enum BuiltinPipePlugin { IOTDB_THRIFT_SYNC_SINK("iotdb-thrift-sync-sink", IoTDBThriftSyncSink.class), IOTDB_THRIFT_ASYNC_SINK("iotdb-thrift-async-sink", IoTDBThriftAsyncSink.class), IOTDB_LEGACY_PIPE_SINK("iotdb-legacy-pipe-sink", IoTDBLegacyPipeSink.class), - IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapSink.class), WEBSOCKET_SINK("websocket-sink", WebSocketSink.class), - OPC_UA_SINK("opc-ua-sink", OpcUaSink.class), - OPC_DA_SINK("opc-da-sink", OpcDaSink.class), WRITE_BACK_SINK("write-back-sink", WriteBackSink.class), SUBSCRIPTION_SINK("subscription-sink", DoNothingSink.class), IOT_CONSENSUS_V2_ASYNC_SINK("iot-consensus-v2-async-sink", IoTConsensusV2AsyncSink.class), @@ -155,14 +128,6 @@ public String getClassName() { // Sources DO_NOTHING_SOURCE.getPipePluginName().toUpperCase(), // Processors - TUMBLING_TIME_SAMPLING_PROCESSOR.getPipePluginName().toUpperCase(), - SDT_SAMPLING_PROCESSOR.getPipePluginName().toUpperCase(), - CHANGING_VALUE_SAMPLING_PROCESSOR.getPipePluginName().toUpperCase(), - THROWING_EXCEPTION_PROCESSOR.getPipePluginName().toUpperCase(), - AGGREGATE_PROCESSOR.getPipePluginName().toUpperCase(), - COUNT_POINT_PROCESSOR.getPipePluginName().toUpperCase(), - STANDARD_STATISTICS_PROCESSOR.getPipePluginName().toUpperCase(), - TUMBLING_WINDOWING_PROCESSOR.getPipePluginName().toUpperCase(), IOT_CONSENSUS_V2_PROCESSOR.getPipePluginName().toUpperCase(), PIPE_CONSENSUS_PROCESSOR.getPipePluginName().toUpperCase(), RENAME_DATABASE_PROCESSOR.getPipePluginName().toUpperCase(), @@ -173,10 +138,7 @@ public String getClassName() { IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName().toUpperCase(), IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(), IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName().toUpperCase(), - IOTDB_AIR_GAP_CONNECTOR.getPipePluginName().toUpperCase(), WEBSOCKET_CONNECTOR.getPipePluginName().toUpperCase(), - OPC_UA_CONNECTOR.getPipePluginName().toUpperCase(), - OPC_DA_CONNECTOR.getPipePluginName().toUpperCase(), WRITE_BACK_CONNECTOR.getPipePluginName().toUpperCase(), IOT_CONSENSUS_V2_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(), PIPE_CONSENSUS_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(), @@ -185,8 +147,6 @@ public String getClassName() { IOTDB_THRIFT_ASYNC_SINK.getPipePluginName().toUpperCase(), IOTDB_LEGACY_PIPE_SINK.getPipePluginName().toUpperCase(), WEBSOCKET_SINK.getPipePluginName().toUpperCase(), - OPC_UA_SINK.getPipePluginName().toUpperCase(), - OPC_DA_SINK.getPipePluginName().toUpperCase(), SUBSCRIPTION_SINK.getPipePluginName().toUpperCase(), IOT_CONSENSUS_V2_ASYNC_SINK.getPipePluginName().toUpperCase(), PIPE_CONSENSUS_ASYNC_SINK.getPipePluginName().toUpperCase()))); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java index aa09b712d6e1c..2a6882789d445 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java @@ -68,13 +68,6 @@ public class PipeSinkConstant { public static final String SINK_IOTDB_PARALLEL_TASKS_KEY = "sink.parallel.tasks"; public static final int CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE = PipeConfig.getInstance().getPipeSubtaskExecutorMaxThreadNum(); - public static final Set SINGLE_THREAD_DEFAULT_SINK = - new HashSet<>( - Arrays.asList( - BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), - BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), - BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(), - BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName())); public static final String CONNECTOR_REALTIME_FIRST_KEY = "connector.realtime-first"; public static final String SINK_REALTIME_FIRST_KEY = "sink.realtime-first"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java index 6315ae53b7f91..54b93ad45c90a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java @@ -69,7 +69,7 @@ public abstract class IoTDBSource implements PipeExtractor { protected int regionId; protected PipeTaskMeta pipeTaskMeta; - protected boolean isForwardingPipeRequests; + protected boolean isForwardingPipeRequests = true; // The value is always true after the first start even the extractor is closed protected final AtomicBoolean hasBeenStarted = new AtomicBoolean(false); @@ -144,15 +144,7 @@ public void customize( pipeTaskMeta = ((PipeTaskSourceRuntimeEnvironment) environment).getPipeTaskMeta(); } - final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters); - if (isDoubleLiving) { - isForwardingPipeRequests = false; - } else { - isForwardingPipeRequests = - parameters.getBooleanOrDefault( - PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS, - PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE); - } + isForwardingPipeRequests = true; userId = parameters.getStringOrDefault( diff --git a/pom.xml b/pom.xml index 4b0f44bbad353..4636dbeea8026 100644 --- a/pom.xml +++ b/pom.xml @@ -744,6 +744,10 @@ 3.1.2 ${argLine} -Xmx1024m + + + **/subscription/** + ${project.build.directory}/fork_${surefire.forkNumber}