This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e231bbe5e2a Fix pipe double-living sugar after isolation (#18119)
e231bbe5e2a is described below
commit e231bbe5e2a2e4cec9e9a4de065910ba447c661b
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 8 17:45:00 2026 +0800
Fix pipe double-living sugar after isolation (#18119)
---
.../manual/enhanced/IoTDBPipeDoubleLivingIT.java | 10 +++---
.../pipe/coordinator/task/PipeTaskCoordinator.java | 30 +++++++++++-----
...istoricalDataRegionTsFileAndDeletionSource.java | 11 ++----
.../realtime/PipeRealtimeDataRegionSource.java | 11 ++----
.../config/executor/ClusterConfigTaskExecutor.java | 25 +++++++++----
.../pipe/config/constant/PipeSourceConstant.java | 41 ++++++++++++++++++++++
.../datastructure/visibility/VisibilityUtils.java | 9 +----
.../iotdb/commons/pipe/source/IoTDBSource.java | 18 ++--------
.../visibility/VisibilityUtilsTest.java | 24 +++++++++++++
9 files changed, 117 insertions(+), 62 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java
index 26f7dd9db9c..8497bbdcf0f 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java
@@ -71,7 +71,7 @@ public class IoTDBPipeDoubleLivingIT extends
AbstractPipeTableModelDualManualIT
"create pipe %s"
+ " with source ("
+ "'capture.tree'='false',"
- + "'mode.double-living'='true')"
+ + "'double-living'='true')"
+ " with sink ("
+ "'node-urls'='%s')",
"p1", receiverDataNode.getIpAndPortString()));
@@ -84,7 +84,7 @@ public class IoTDBPipeDoubleLivingIT extends
AbstractPipeTableModelDualManualIT
"create pipe %s"
+ " with source ("
+ "'capture.table'='false',"
- + "'mode.double-living'='true')"
+ + "'source.mode.double-living'='true')"
+ " with sink ("
+ "'node-urls'='%s')",
"p2", receiverDataNode.getIpAndPortString()));
@@ -286,7 +286,7 @@ public class IoTDBPipeDoubleLivingIT extends
AbstractPipeTableModelDualManualIT
String.format(
"create pipe %s"
+ " with source ("
- + "'mode.double-living'='true')"
+ + "'double-living'='true')"
+ " with sink ("
+ "'node-urls'='%s')",
treePipeName, receiverDataNode.getIpAndPortString()));
@@ -308,7 +308,7 @@ public class IoTDBPipeDoubleLivingIT extends
AbstractPipeTableModelDualManualIT
String.format(
"create pipe %s"
+ " with source ("
- + "'mode.double-living'='true')"
+ + "'source.mode.double-living'='true')"
+ " with sink ("
+ "'node-urls'='%s')",
tablePipeName, receiverDataNode.getIpAndPortString()));
@@ -350,7 +350,7 @@ public class IoTDBPipeDoubleLivingIT extends
AbstractPipeTableModelDualManualIT
statement.execute(
String.format(
"create pipe %s"
- + " with source ('mode.double-living'='true')"
+ + " with source ('extractor.double-living'='true')"
+ " with sink ('node-urls'='%s')",
pipeName, receiverDataNode.getIpAndPortString()));
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
index 930bf6cadf3..42d61750c9b 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.confignode.manager.pipe.coordinator.task;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.i18n.PipeMessages;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
@@ -48,7 +49,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -121,16 +121,15 @@ public class PipeTaskCoordinator {
}
private boolean isDoubleLivingPipe(final TCreatePipeReq req) {
- return new PipeParameters(req.getExtractorAttributes())
- .getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ return PipeSourceConstant.isDoubleLiving(new
PipeParameters(req.getExtractorAttributes()));
}
private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) {
final PipeParameters sourceParameters = new
PipeParameters(req.getExtractorAttributes());
+ final TSStatus validationStatus =
validateDoubleLivingPipeParameters(sourceParameters);
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
validationStatus.getCode()) {
+ return validationStatus;
+ }
final String currentDialect =
sourceParameters.getStringOrDefault(
SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TREE_VALUE);
@@ -180,11 +179,24 @@ public class PipeTaskCoordinator {
return secondStatus;
}
+ private TSStatus validateDoubleLivingPipeParameters(final PipeParameters
sourceParameters) {
+ final Boolean isForwardingPipeRequests =
+ sourceParameters.getBooleanByKeys(
+ PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
+ PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY);
+ return Boolean.TRUE.equals(isForwardingPipeRequests)
+ ? RpcUtils.getStatus(
+ TSStatusCode.PIPE_ERROR,
+ PipeMessages
+
.EXCEPTION_FORWARDING_PIPE_REQUESTS_CAN_NOT_SPECIFIED_TRUE_DOUBLE_LIVING_ENABLED_B000E8A1)
+ : RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ }
+
private TCreatePipeReq cloneCreatePipeRequestWithDialect(
final TCreatePipeReq req, final PipeParameters sourceParameters, final
String sqlDialect) {
final Map<String, String> sourceAttributes = new
HashMap<>(sourceParameters.getAttribute());
-
sourceAttributes.remove(PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY);
- sourceAttributes.remove(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
+ PipeSourceConstant.removeDoubleLivingAttributes(sourceAttributes);
+ PipeSourceConstant.disableForwardingPipeRequests(sourceAttributes);
sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
sourceAttributes.put(
SystemConstant.PIPE_VISIBILITY_KEY,
SystemConstant.PIPE_VISIBILITY_STRICT_VALUE);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
index aea86f85dda..48057cb687b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
@@ -395,20 +395,13 @@ public class
PipeHistoricalDataRegionTsFileAndDeletionSource
skipIfNoPrivileges = getSkipIfNoPrivileges(parameters);
- final boolean isDoubleLiving =
- parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ final boolean isDoubleLiving =
PipeSourceConstant.isDoubleLiving(parameters);
if (isDoubleLiving) {
isForwardingPipeRequests = false;
} else {
isForwardingPipeRequests =
parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
- PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+ PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
}
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 0d03df454ed..732275824eb 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
@@ -262,20 +262,13 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
?
TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime)
:
TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1;
- final boolean isDoubleLiving =
- parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ final boolean isDoubleLiving =
PipeSourceConstant.isDoubleLiving(parameters);
if (isDoubleLiving) {
isForwardingPipeRequests = false;
} else {
isForwardingPipeRequests =
parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
- PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+ PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 92b9b80f1c6..3855556e6bd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -60,6 +60,7 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.exception.SemanticException;
import org.apache.iotdb.commons.executable.ExecutableManager;
import org.apache.iotdb.commons.executable.ExecutableResource;
+import org.apache.iotdb.commons.i18n.PipeMessages;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
@@ -347,6 +348,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.Compacti
import org.apache.iotdb.db.trigger.service.TriggerClassLoader;
import org.apache.iotdb.pipe.api.PipePlugin;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
import org.apache.iotdb.pipe.api.exception.PipePasswordCheckException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
@@ -2278,6 +2280,7 @@ public class ClusterConfigTaskExecutor implements
IConfigTaskExecutor {
// Validate pipe plugin before creation
try {
if (isDoubleLivingPipe(sourcePipeParameters)) {
+ validateDoubleLivingPipeParameters(sourcePipeParameters);
validatePipePlugin(
pipeName,
cloneSourceParametersWithDialect(
@@ -2342,18 +2345,26 @@ public class ClusterConfigTaskExecutor implements
IConfigTaskExecutor {
}
private boolean isDoubleLivingPipe(final PipeParameters
sourcePipeParameters) {
- return sourcePipeParameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ return PipeSourceConstant.isDoubleLiving(sourcePipeParameters);
+ }
+
+ private void validateDoubleLivingPipeParameters(final PipeParameters
sourcePipeParameters) {
+ final Boolean isForwardingPipeRequests =
+ sourcePipeParameters.getBooleanByKeys(
+ PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
+ PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY);
+ if (Boolean.TRUE.equals(isForwardingPipeRequests)) {
+ throw new PipeParameterNotValidException(
+ PipeMessages
+
.EXCEPTION_FORWARDING_PIPE_REQUESTS_CAN_NOT_SPECIFIED_TRUE_DOUBLE_LIVING_ENABLED_B000E8A1);
+ }
}
private PipeParameters cloneSourceParametersWithDialect(
final PipeParameters sourcePipeParameters, final String sqlDialect) {
final Map<String, String> sourceAttributes = new
HashMap<>(sourcePipeParameters.getAttribute());
-
sourceAttributes.remove(PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY);
- sourceAttributes.remove(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
+ PipeSourceConstant.removeDoubleLivingAttributes(sourceAttributes);
+ PipeSourceConstant.disableForwardingPipeRequests(sourceAttributes);
sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
sourceAttributes.put(
SystemConstant.PIPE_VISIBILITY_KEY,
SystemConstant.PIPE_VISIBILITY_STRICT_VALUE);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
index e755ef3c601..28af96637f1 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
@@ -20,10 +20,13 @@
package org.apache.iotdb.commons.pipe.config.constant;
import org.apache.iotdb.commons.i18n.PipeMessages;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
public class PipeSourceConstant {
@@ -87,6 +90,10 @@ public class PipeSourceConstant {
public static final String SOURCE_FORWARDING_PIPE_REQUESTS_KEY =
"source.forwarding-pipe-requests";
public static final boolean EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE
= true;
+ public static final List<String> FORWARDING_PIPE_REQUESTS_KEYS =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
SOURCE_FORWARDING_PIPE_REQUESTS_KEY));
public static final String EXTRACTOR_HISTORY_ENABLE_KEY =
"extractor.history.enable";
public static final String SOURCE_HISTORY_ENABLE_KEY =
"source.history.enable";
@@ -137,7 +144,16 @@ public class PipeSourceConstant {
public static final boolean EXTRACTOR_MODE_SNAPSHOT_DEFAULT_VALUE = false;
public static final String EXTRACTOR_MODE_DOUBLE_LIVING_KEY =
"extractor.mode.double-living";
public static final String SOURCE_MODE_DOUBLE_LIVING_KEY =
"source.mode.double-living";
+ public static final String EXTRACTOR_DOUBLE_LIVING_KEY =
"extractor.double-living";
+ public static final String SOURCE_DOUBLE_LIVING_KEY = "source.double-living";
public static final boolean EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE =
false;
+ public static final List<String> DOUBLE_LIVING_KEYS =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
+ SOURCE_MODE_DOUBLE_LIVING_KEY,
+ EXTRACTOR_DOUBLE_LIVING_KEY,
+ SOURCE_DOUBLE_LIVING_KEY));
public static final String EXTRACTOR_START_TIME_KEY = "extractor.start-time";
public static final String SOURCE_START_TIME_KEY = "source.start-time";
@@ -195,6 +211,31 @@ public class PipeSourceConstant {
public static final String EXTRACTOR_CONSENSUS_RECEIVER_DATANODE_ID_KEY =
"extractor.consensus.receiver-dn-id";
+ public static boolean isDoubleLiving(final PipeParameters sourceParameters) {
+ return sourceParameters.getBooleanOrDefault(
+ DOUBLE_LIVING_KEYS, EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ }
+
+ public static void removeDoubleLivingAttributes(final Map<String, String>
sourceAttributes) {
+ removeEquivalentAttributes(sourceAttributes, DOUBLE_LIVING_KEYS);
+ }
+
+ public static void disableForwardingPipeRequests(final Map<String, String>
sourceAttributes) {
+ removeEquivalentAttributes(sourceAttributes,
FORWARDING_PIPE_REQUESTS_KEYS);
+ sourceAttributes.put(SOURCE_FORWARDING_PIPE_REQUESTS_KEY,
Boolean.FALSE.toString());
+ }
+
+ private static void removeEquivalentAttributes(
+ final Map<String, String> sourceAttributes, final List<String>
equivalentKeys) {
+ final Set<String> reducedKeys =
+ equivalentKeys.stream()
+ .map(PipeParameters.KeyReducer::reduce)
+ .collect(java.util.stream.Collectors.toSet());
+ sourceAttributes
+ .keySet()
+ .removeIf(key ->
reducedKeys.contains(PipeParameters.KeyReducer.reduce(key)));
+ }
+
private PipeSourceConstant() {
throw new IllegalStateException(PipeMessages.UTILITY_CLASS);
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
index 3d45715f9aa..d01803b8950 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
@@ -27,16 +27,10 @@ import
org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import java.util.Arrays;
-import java.util.List;
import java.util.Objects;
public class VisibilityUtils {
- private static final List<String> DOUBLE_LIVING_KEYS =
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
-
private VisibilityUtils() {
// forbidding instantiation
}
@@ -91,8 +85,7 @@ public class VisibilityUtils {
return calculateStrictlyFromExtractorParameters(extractorParameters);
}
- if (extractorParameters.getBooleanOrDefault(
- DOUBLE_LIVING_KEYS,
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE)) {
+ if (PipeSourceConstant.isDoubleLiving(extractorParameters)) {
return Visibility.BOTH;
}
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 c2dd6855d01..6315ae53b7f 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
@@ -115,12 +115,7 @@ public abstract class IoTDBSource implements PipeExtractor
{
}
private void validateDoubleLiving(final PipeParameters parameters) {
- final boolean isDoubleLiving =
- parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ final boolean isDoubleLiving =
PipeSourceConstant.isDoubleLiving(parameters);
if (!isDoubleLiving) {
return;
}
@@ -149,20 +144,13 @@ public abstract class IoTDBSource implements
PipeExtractor {
pipeTaskMeta = ((PipeTaskSourceRuntimeEnvironment)
environment).getPipeTaskMeta();
}
- final boolean isDoubleLiving =
- parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ final boolean isDoubleLiving =
PipeSourceConstant.isDoubleLiving(parameters);
if (isDoubleLiving) {
isForwardingPipeRequests = false;
} else {
isForwardingPipeRequests =
parameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
- PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+ PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java
index 4cec5572dd7..68d5e9262b9 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java
@@ -70,6 +70,10 @@ public class VisibilityUtilsTest {
doubleLivingAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY,
"false");
assertVisibility(Visibility.BOTH, doubleLivingAttributes, true, true);
+ final Map<String, String> directDoubleLivingAttributes = new HashMap<>();
+ directDoubleLivingAttributes.put("double-living", "true");
+ assertVisibility(Visibility.BOTH, directDoubleLivingAttributes, true,
true);
+
final Map<String, String> captureBothAttributes = new HashMap<>();
captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY,
"true");
captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY,
"true");
@@ -81,6 +85,26 @@ public class VisibilityUtilsTest {
assertVisibility(Visibility.NONE, captureNoneAttributes, false, false);
}
+ @Test
+ public void testDoubleLivingSugarCanBeNormalizedForStrictPipes() {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("double-living", "true");
+ attributes.put("mode.double-living", "true");
+ attributes.put(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY, "true");
+ attributes.put("forwarding-pipe-requests", "true");
+ attributes.put(PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
"true");
+
+ PipeSourceConstant.removeDoubleLivingAttributes(attributes);
+ PipeSourceConstant.disableForwardingPipeRequests(attributes);
+
+ final PipeParameters parameters = new PipeParameters(attributes);
+ Assert.assertFalse(PipeSourceConstant.isDoubleLiving(parameters));
+ Assert.assertFalse(
+ parameters.getBooleanOrDefault(
+ PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
+
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE));
+ }
+
private void assertVisibility(
final Visibility expectedVisibility,
final Map<String, String> attributes,
