You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/11/15 08:02:59 UTC
(iotdb) branch rc/1.3.3 updated: Pipe: Disable unstable features in the distribution
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/1.3.3 by this push:
new 2a8bb0253be Pipe: Disable unstable features in the distribution
2a8bb0253be is described below
commit 2a8bb0253bee39c3ba3fa028e051b281d10d2bcc
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Nov 15 15:55:32 2024 +0800
Pipe: Disable unstable features in the distribution
---
.github/workflows/pipe-it-2cluster.yml | 230 ---------------------
.../pipe/it/autocreate/IoTDBPipeProtocolIT.java | 14 --
.../PipeConfigRegionConnectorConstructor.java | 7 -
.../PipeDataRegionConnectorConstructor.java | 11 -
.../PipeDataRegionProcessorConstructor.java | 30 ---
.../PipeSchemaRegionConnectorConstructor.java | 7 -
.../dataregion/IoTDBDataRegionExtractor.java | 77 +------
.../realtime/PipeRealtimeDataRegionExtractor.java | 10 +-
.../agent/plugin/builtin/BuiltinPipePlugin.java | 35 ----
.../commons/pipe/extractor/IoTDBExtractor.java | 10 +-
10 files changed, 6 insertions(+), 425 deletions(-)
diff --git a/.github/workflows/pipe-it-2cluster.yml b/.github/workflows/pipe-it-2cluster.yml
deleted file mode 100644
index 83a994e8e33..00000000000
--- a/.github/workflows/pipe-it-2cluster.yml
+++ /dev/null
@@ -1,230 +0,0 @@
-name: Multi-Cluster IT
-
-on:
- push:
- branches:
- - master
- - 'rel/1.*'
- - 'rc/1.*'
- paths-ignore:
- - 'docs/**'
- - 'site/**'
- pull_request:
- branches:
- - master
- - 'rel/1.*'
- - 'rc/1.*'
- paths-ignore:
- - 'docs/**'
- - 'site/**'
- # 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.GE_ACCESS_TOKEN }}
-
-jobs:
- auto-create-schema:
- 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]
- os: [ ubuntu-latest ]
- runs-on: ${{ matrix.os }}
- steps:
- - uses: actions/checkout@v4
- - name: Set up JDK ${{ matrix.java }}
- uses: actions/setup-java@v4
- with:
- distribution: liberica
- java-version: ${{ matrix.java }}
- - 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: |
- 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 -PMultiClusterIT2AutoCreateSchema \
- -ntp
- - name: Upload Artifact
- if: failure()
- uses: actions/upload-artifact@v4
- with:
- name: cluster-log-auto-create-schema-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}
- path: integration-test/target/cluster-logs
- retention-days: 30
- manual-create-schema:
- 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]
- cluster2: [LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode]
- os: [ ubuntu-latest ]
- exclude:
- - cluster1: LightWeightStandaloneMode
- cluster2: LightWeightStandaloneMode
- - cluster1: LightWeightStandaloneMode
- cluster2: ScalableSingleNodeMode
- - cluster1: ScalableSingleNodeMode
- cluster2: LightWeightStandaloneMode
- - cluster1: HighPerformanceMode
- cluster2: LightWeightStandaloneMode
- runs-on: ${{ matrix.os }}
- steps:
- - uses: actions/checkout@v4
- - name: Set up JDK ${{ matrix.java }}
- uses: actions/setup-java@v4
- with:
- distribution: liberica
- java-version: ${{ matrix.java }}
- - 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: |
- 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 -PMultiClusterIT2ManualCreateSchema \
- -ntp
- - name: Upload Artifact
- if: failure()
- uses: actions/upload-artifact@v4
- with:
- name: cluster-log-manual-create-schema-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}
- path: integration-test/target/cluster-logs
- retention-days: 30
- subscription-arch-verification:
- 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 ]
- cluster2: [ ScalableSingleNodeMode ]
- os: [ ubuntu-latest ]
- runs-on: ${{ matrix.os }}
- steps:
- - uses: actions/checkout@v4
- - name: Set up JDK ${{ matrix.java }}
- uses: actions/setup-java@v4
- with:
- distribution: liberica
- java-version: ${{ matrix.java }}
- - 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: |
- 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 -PMultiClusterIT2SubscriptionArchVerification \
- -ntp
- - name: Upload Artifact
- if: failure()
- uses: actions/upload-artifact@v4
- with:
- name: cluster-log-subscription-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}
- path: integration-test/target/cluster-logs
- retention-days: 30
- subscription-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@v4
- - name: Set up JDK ${{ matrix.java }}
- uses: actions/setup-java@v4
- with:
- distribution: liberica
- java-version: ${{ matrix.java }}
- - 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: |
- 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 -PMultiClusterIT2SubscriptionRegressionConsumer \
- -ntp
- - name: Upload Artifact
- if: failure()
- uses: actions/upload-artifact@v4
- with:
- name: cluster-log-subscription-regression-consumer-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}
- path: integration-test/target/cluster-logs
- retention-days: 30
- subscription-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@v4
- - name: Set up JDK ${{ matrix.java }}
- uses: actions/setup-java@v4
- with:
- distribution: liberica
- java-version: ${{ matrix.java }}
- - 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: |
- 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 -PMultiClusterIT2SubscriptionRegressionMisc \
- -ntp
- - name: Upload Artifact
- if: failure()
- uses: actions/upload-artifact@v4
- with:
- name: cluster-log-subscription-regression-misc-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}
- path: integration-test/target/cluster-logs
- retention-days: 30
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java
index 328e1af5fd7..20e8e6317d5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java
@@ -342,11 +342,6 @@ public class IoTDBPipeProtocolIT extends AbstractPipeDualAutoIT {
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()
@@ -381,16 +376,7 @@ public class IoTDBPipeProtocolIT extends AbstractPipeDualAutoIT {
final StringBuilder nodeUrlsBuilder = new StringBuilder();
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(",");
- }
}
try (final SyncConfigNodeIServiceClient client =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionConnectorConstructor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionConnectorConstructor.java
index 74a34e503be..514c4aa7d50 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionConnectorConstructor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionConnectorConstructor.java
@@ -22,7 +22,6 @@ package org.apache.iotdb.confignode.manager.pipe.agent.plugin;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.donothing.DoNothingConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeConnectorConstructor;
-import org.apache.iotdb.confignode.manager.pipe.connector.protocol.IoTDBConfigRegionAirGapConnector;
import org.apache.iotdb.confignode.manager.pipe.connector.protocol.IoTDBConfigRegionConnector;
import org.apache.iotdb.pipe.api.PipeConnector;
@@ -42,9 +41,6 @@ class PipeConfigRegionConnectorConstructor extends PipeConnectorConstructor {
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
IoTDBConfigRegionConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(),
- IoTDBConfigRegionAirGapConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingConnector::new);
@@ -59,9 +55,6 @@ class PipeConfigRegionConnectorConstructor extends PipeConnectorConstructor {
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(),
IoTDBConfigRegionConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(),
- IoTDBConfigRegionAirGapConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingConnector::new);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionConnectorConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionConnectorConstructor.java
index e11e8dc5685..ea358383835 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionConnectorConstructor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionConnectorConstructor.java
@@ -23,9 +23,7 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.donothing.DoNothingConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeConnectorConstructor;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper;
-import org.apache.iotdb.db.pipe.connector.protocol.airgap.IoTDBDataRegionAirGapConnector;
import org.apache.iotdb.db.pipe.connector.protocol.legacy.IoTDBLegacyPipeConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.opcua.OpcUaConnector;
import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.PipeConsensusAsyncConnector;
import org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBDataRegionAsyncConnector;
import org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBDataRegionSyncConnector;
@@ -58,13 +56,8 @@ class PipeDataRegionConnectorConstructor extends PipeConnectorConstructor {
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName(),
IoTDBLegacyPipeConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(),
- IoTDBDataRegionAirGapConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(), WebSocketConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), OpcUaConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingConnector::new);
pluginConstructors.put(
@@ -85,12 +78,8 @@ class PipeDataRegionConnectorConstructor extends PipeConnectorConstructor {
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_SINK.getPipePluginName(),
IoTDBLegacyPipeConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(),
- IoTDBDataRegionAirGapConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(), WebSocketConnector::new);
- pluginConstructors.put(BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), OpcUaConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingConnector::new);
pluginConstructors.put(
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 44c6ef17800..9e0ba96d197 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,17 +21,9 @@ package org.apache.iotdb.db.pipe.agent.plugin.dataregion;
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.pipeconsensus.PipeConsensusProcessor;
-import org.apache.iotdb.db.pipe.processor.twostage.plugin.TwoStageCountProcessor;
class PipeDataRegionProcessorConstructor extends PipeProcessorConstructor {
@@ -43,28 +35,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.PIPE_CONSENSUS_PROCESSOR.getPipePluginName(),
PipeConsensusProcessor::new);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionConnectorConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionConnectorConstructor.java
index 6c96aaaeddc..fb87b90f877 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionConnectorConstructor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionConnectorConstructor.java
@@ -22,7 +22,6 @@ package org.apache.iotdb.db.pipe.agent.plugin.schemaregion;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.donothing.DoNothingConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeConnectorConstructor;
-import org.apache.iotdb.db.pipe.connector.protocol.airgap.IoTDBSchemaRegionAirGapConnector;
import org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBSchemaRegionConnector;
import org.apache.iotdb.pipe.api.PipeConnector;
@@ -42,9 +41,6 @@ class PipeSchemaRegionConnectorConstructor extends PipeConnectorConstructor {
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
IoTDBSchemaRegionConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(),
- IoTDBSchemaRegionAirGapConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingConnector::new);
@@ -59,9 +55,6 @@ class PipeSchemaRegionConnectorConstructor extends PipeConnectorConstructor {
pluginConstructors.put(
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(),
IoTDBSchemaRegionConnector::new);
- pluginConstructors.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(),
- IoTDBSchemaRegionAirGapConnector::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingConnector::new);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
index a2d91be2ab8..56c688501b0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.db.pipe.extractor.dataregion;
import org.apache.iotdb.commons.consensus.DataRegionId;
-import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.extractor.IoTDBExtractor;
@@ -36,7 +35,6 @@ import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.PipeRealtimeDataRe
import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.PipeRealtimeDataRegionTsFileExtractor;
import org.apache.iotdb.db.pipe.metric.PipeDataRegionExtractorMetrics;
import org.apache.iotdb.db.storageengine.StorageEngine;
-import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -67,13 +65,6 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_PATTERN_FORMAT_PREFIX_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_ENABLE_KEY;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_BATCH_MODE_VALUE;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_FILE_VALUE;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_FORCED_LOG_VALUE;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_HYBRID_VALUE;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_KEY;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_LOG_VALUE;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_STREAM_MODE_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_START_TIME_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_WATERMARK_INTERVAL_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_WATERMARK_INTERVAL_KEY;
@@ -84,7 +75,6 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_MODE_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_PATTERN_FORMAT_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_REALTIME_ENABLE_KEY;
-import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_REALTIME_MODE_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_START_TIME_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_WATERMARK_INTERVAL_KEY;
@@ -169,26 +159,6 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY),
EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE));
- // Validate extractor.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, SOURCE_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);
- }
-
// Validate source.start-time and source.end-time
if (validator.getParameters().hasAnyAttributes(SOURCE_START_TIME_KEY, SOURCE_END_TIME_KEY)
&& validator
@@ -238,8 +208,7 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
historicalExtractor = new PipeHistoricalDataRegionTsFileExtractor();
}
- private void constructRealtimeExtractor(final PipeParameters parameters)
- throws IllegalPathException {
+ private void constructRealtimeExtractor(final PipeParameters parameters) {
// Use heartbeat only extractor if disable realtime extractor
if (!parameters.getBooleanOrDefault(
Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY),
@@ -265,49 +234,7 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
return;
}
- // Use hybrid mode by default
- if (!parameters.hasAnyAttributes(EXTRACTOR_REALTIME_MODE_KEY, SOURCE_REALTIME_MODE_KEY)) {
- checkWalEnable(parameters);
- realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
- LOGGER.info(
- "Pipe: '{}' is not set, use hybrid mode by default.", EXTRACTOR_REALTIME_MODE_KEY);
- return;
- }
-
- switch (parameters.getStringByKeys(EXTRACTOR_REALTIME_MODE_KEY, SOURCE_REALTIME_MODE_KEY)) {
- case EXTRACTOR_REALTIME_MODE_FILE_VALUE:
- case EXTRACTOR_REALTIME_MODE_BATCH_MODE_VALUE:
- realtimeExtractor = new PipeRealtimeDataRegionTsFileExtractor();
- break;
- case EXTRACTOR_REALTIME_MODE_HYBRID_VALUE:
- case EXTRACTOR_REALTIME_MODE_LOG_VALUE:
- case EXTRACTOR_REALTIME_MODE_STREAM_MODE_VALUE:
- checkWalEnable(parameters);
- realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
- break;
- case EXTRACTOR_REALTIME_MODE_FORCED_LOG_VALUE:
- checkWalEnable(parameters);
- realtimeExtractor = new PipeRealtimeDataRegionLogExtractor();
- break;
- default:
- checkWalEnable(parameters);
- realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
- if (LOGGER.isWarnEnabled()) {
- LOGGER.warn(
- "Pipe: Unsupported extractor realtime mode: {}, create a hybrid extractor.",
- parameters.getStringByKeys(EXTRACTOR_REALTIME_MODE_KEY, SOURCE_REALTIME_MODE_KEY));
- }
- }
- }
-
- private void checkWalEnable(final PipeParameters parameters) throws IllegalPathException {
- if (Boolean.TRUE.equals(
- DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(parameters)
- .getLeft())
- && IoTDBDescriptor.getInstance().getConfig().getWalMode().equals(WALMode.DISABLE)) {
- throw new PipeException(
- "The pipe cannot transfer realtime insertion if data region disables wal. Please set 'realtime.mode'='batch' in source parameters when enabling realtime transmission.");
- }
+ realtimeExtractor = new PipeRealtimeDataRegionTsFileExtractor();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java
index 40c6bdc906f..7988606ad31 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -103,7 +102,7 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
private final AtomicReference<Pair<Long, Long>> dataRegionTimePartitionIdBound =
new AtomicReference<>();
- protected boolean isForwardingPipeRequests;
+ protected boolean isForwardingPipeRequests = true;
private boolean shouldTransferModFile; // Whether to transfer mods
@@ -226,12 +225,7 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime)
: TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1;
- isForwardingPipeRequests =
- parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeExtractorConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
- PipeExtractorConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
- PipeExtractorConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
+ isForwardingPipeRequests = true;
shouldTransferModFile =
parameters.getBooleanOrDefault(
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 67f3a0265af..6fd03eac63f 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
@@ -20,28 +20,18 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.donothing.DoNothingConnector;
-import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.airgap.IoTDBAirGapConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.consensus.PipeConsensusAsyncConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift.IoTDBLegacyPipeConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift.IoTDBThriftAsyncConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift.IoTDBThriftConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift.IoTDBThriftSslConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift.IoTDBThriftSyncConnector;
-import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.opcua.OpcUaConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.websocket.WebSocketConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.writeback.WriteBackConnector;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.extractor.donothing.DoNothingExtractor;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.extractor.iotdb.IoTDBExtractor;
-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.pipeconsensus.PipeConsensusProcessor;
-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 java.util.Arrays;
import java.util.Collections;
@@ -59,18 +49,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),
PIPE_CONSENSUS_PROCESSOR("pipe-consensus-processor", PipeConsensusProcessor.class),
// connectors
@@ -80,12 +60,10 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_SYNC_CONNECTOR("iotdb-thrift-sync-connector", IoTDBThriftSyncConnector.class),
IOTDB_THRIFT_ASYNC_CONNECTOR("iotdb-thrift-async-connector", IoTDBThriftAsyncConnector.class),
IOTDB_LEGACY_PIPE_CONNECTOR("iotdb-legacy-pipe-connector", IoTDBLegacyPipeConnector.class),
- IOTDB_AIR_GAP_CONNECTOR("iotdb-air-gap-connector", IoTDBAirGapConnector.class),
PIPE_CONSENSUS_ASYNC_CONNECTOR(
"pipe-consensus-async-connector", PipeConsensusAsyncConnector.class),
WEBSOCKET_CONNECTOR("websocket-connector", WebSocketConnector.class),
- OPC_UA_CONNECTOR("opc-ua-connector", OpcUaConnector.class),
WRITE_BACK_CONNECTOR("write-back-connector", WriteBackConnector.class),
DO_NOTHING_SINK("do-nothing-sink", DoNothingConnector.class),
@@ -94,9 +72,7 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_SYNC_SINK("iotdb-thrift-sync-sink", IoTDBThriftSyncConnector.class),
IOTDB_THRIFT_ASYNC_SINK("iotdb-thrift-async-sink", IoTDBThriftAsyncConnector.class),
IOTDB_LEGACY_PIPE_SINK("iotdb-legacy-pipe-sink", IoTDBLegacyPipeConnector.class),
- IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapConnector.class),
WEBSOCKET_SINK("websocket-sink", WebSocketConnector.class),
- OPC_UA_SINK("opc-ua-sink", OpcUaConnector.class),
WRITE_BACK_SINK("write-back-sink", WriteBackConnector.class),
SUBSCRIPTION_SINK("subscription-sink", DoNothingConnector.class),
PIPE_CONSENSUS_ASYNC_SINK("pipe-consensus-async-sink", PipeConsensusAsyncConnector.class),
@@ -134,14 +110,6 @@ public enum BuiltinPipePlugin {
// 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(),
PIPE_CONSENSUS_PROCESSOR.getPipePluginName().toUpperCase(),
// Connectors
DO_NOTHING_CONNECTOR.getPipePluginName().toUpperCase(),
@@ -150,9 +118,7 @@ public enum BuiltinPipePlugin {
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(),
WRITE_BACK_CONNECTOR.getPipePluginName().toUpperCase(),
PIPE_CONSENSUS_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(),
// Sinks
@@ -160,7 +126,6 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_ASYNC_SINK.getPipePluginName().toUpperCase(),
IOTDB_LEGACY_PIPE_SINK.getPipePluginName().toUpperCase(),
WEBSOCKET_SINK.getPipePluginName().toUpperCase(),
- OPC_UA_SINK.getPipePluginName().toUpperCase(),
WRITE_BACK_SINK.getPipePluginName().toUpperCase(),
SUBSCRIPTION_SINK.getPipePluginName().toUpperCase(),
PIPE_CONSENSUS_ASYNC_SINK.getPipePluginName().toUpperCase())));
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java
index 2d3a1615f00..1e7d3eb53c2 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.commons.pipe.extractor;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
import org.apache.iotdb.pipe.api.PipeExtractor;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
@@ -48,7 +47,7 @@ public abstract class IoTDBExtractor 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);
@@ -99,12 +98,7 @@ public abstract class IoTDBExtractor implements PipeExtractor {
taskID = pipeName + "_" + regionId + "_" + creationTime;
pipeTaskMeta = environment.getPipeTaskMeta();
- isForwardingPipeRequests =
- parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeExtractorConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
- PipeExtractorConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
- PipeExtractorConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
+ isForwardingPipeRequests = true;
}
@Override