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 26f7dd9db9ca..8497bbdcf0f7 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 void testDoubleLivingInvalidForwardingParameter() throws Exception { "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 void testDoubleLivingInvalidForwardingParameter() throws Exception { "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 void testDoubleLivingIsolation() throws Exception { 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 void testDoubleLivingIsolation() throws Exception { 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 void testDoubleLivingCreatesSameNamePipesWithSeparateDialects() throws Ex 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 930bf6cadf39..42d61750c9b0 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.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 TSStatus createPipe(TCreatePipeReq req) { } 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 @@ private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) { 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 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 d2f6df290a0b..70cd237099cc 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 void customize( 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 62136877fc1b..c69f3103d965 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 void customize( ? 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 92b9b80f1c6e..3855556e6bd2 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.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.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 SettableFuture createPipe( // Validate pipe plugin before creation try { if (isDoubleLivingPipe(sourcePipeParameters)) { + validateDoubleLivingPipeParameters(sourcePipeParameters); validatePipePlugin( pipeName, cloneSourceParametersWithDialect( @@ -2342,18 +2345,26 @@ private void validatePipePlugin( } 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 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 e755ef3c6019..28af96637f1d 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 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 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 sourceAttributes) { + removeEquivalentAttributes(sourceAttributes, DOUBLE_LIVING_KEYS); + } + + public static void disableForwardingPipeRequests(final Map sourceAttributes) { + removeEquivalentAttributes(sourceAttributes, FORWARDING_PIPE_REQUESTS_KEYS); + sourceAttributes.put(SOURCE_FORWARDING_PIPE_REQUESTS_KEY, Boolean.FALSE.toString()); + } + + private static void removeEquivalentAttributes( + final Map sourceAttributes, final List equivalentKeys) { + final Set 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 3d45715f9aa2..d01803b8950d 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.rpc.subscription.config.TopicConfig; import java.util.Arrays; -import java.util.List; import java.util.Objects; public class VisibilityUtils { - private static final List 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 static Visibility calculateFromExtractorParameters( 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 c2dd6855d010..6315ae53b7f9 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 void validate(final PipeParameterValidator validator) throws Exception { } 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 void customize( 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 4cec5572dd72..68d5e9262b9b 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 void testLegacyVisibilityKeepsDoubleLivingAndCaptureAttributes() { doubleLivingAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY, "false"); assertVisibility(Visibility.BOTH, doubleLivingAttributes, true, true); + final Map directDoubleLivingAttributes = new HashMap<>(); + directDoubleLivingAttributes.put("double-living", "true"); + assertVisibility(Visibility.BOTH, directDoubleLivingAttributes, true, true); + final Map captureBothAttributes = new HashMap<>(); captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY, "true"); captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY, "true"); @@ -81,6 +85,26 @@ public void testLegacyVisibilityKeepsDoubleLivingAndCaptureAttributes() { assertVisibility(Visibility.NONE, captureNoneAttributes, false, false); } + @Test + public void testDoubleLivingSugarCanBeNormalizedForStrictPipes() { + final Map 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 attributes,