Skip to content

Commit e231bbe

Browse files
CaideyipiCaideyipi
authored
Fix pipe double-living sugar after isolation (#18119)
1 parent 7558aa7 commit e231bbe

9 files changed

Lines changed: 117 additions & 62 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ public void testDoubleLivingInvalidForwardingParameter() throws Exception {
7171
"create pipe %s"
7272
+ " with source ("
7373
+ "'capture.tree'='false',"
74-
+ "'mode.double-living'='true')"
74+
+ "'double-living'='true')"
7575
+ " with sink ("
7676
+ "'node-urls'='%s')",
7777
"p1", receiverDataNode.getIpAndPortString()));
@@ -84,7 +84,7 @@ public void testDoubleLivingInvalidForwardingParameter() throws Exception {
8484
"create pipe %s"
8585
+ " with source ("
8686
+ "'capture.table'='false',"
87-
+ "'mode.double-living'='true')"
87+
+ "'source.mode.double-living'='true')"
8888
+ " with sink ("
8989
+ "'node-urls'='%s')",
9090
"p2", receiverDataNode.getIpAndPortString()));
@@ -286,7 +286,7 @@ public void testDoubleLivingIsolation() throws Exception {
286286
String.format(
287287
"create pipe %s"
288288
+ " with source ("
289-
+ "'mode.double-living'='true')"
289+
+ "'double-living'='true')"
290290
+ " with sink ("
291291
+ "'node-urls'='%s')",
292292
treePipeName, receiverDataNode.getIpAndPortString()));
@@ -308,7 +308,7 @@ public void testDoubleLivingIsolation() throws Exception {
308308
String.format(
309309
"create pipe %s"
310310
+ " with source ("
311-
+ "'mode.double-living'='true')"
311+
+ "'source.mode.double-living'='true')"
312312
+ " with sink ("
313313
+ "'node-urls'='%s')",
314314
tablePipeName, receiverDataNode.getIpAndPortString()));
@@ -350,7 +350,7 @@ public void testDoubleLivingCreatesSameNamePipesWithSeparateDialects() throws Ex
350350
statement.execute(
351351
String.format(
352352
"create pipe %s"
353-
+ " with source ('mode.double-living'='true')"
353+
+ " with source ('extractor.double-living'='true')"
354354
+ " with sink ('node-urls'='%s')",
355355
pipeName, receiverDataNode.getIpAndPortString()));
356356
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java‎

Lines changed: 21 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.apache.iotdb.confignode.manager.pipe.coordinator.task;
2121

2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.commons.i18n.PipeMessages;
2324
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
2425
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
2526
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
@@ -48,7 +49,6 @@
4849
import org.slf4j.LoggerFactory;
4950

5051
import java.io.IOException;
51-
import java.util.Arrays;
5252
import java.util.Collections;
5353
import java.util.HashMap;
5454
import java.util.Map;
@@ -121,16 +121,15 @@ public TSStatus createPipe(TCreatePipeReq req) {
121121
}
122122

123123
private boolean isDoubleLivingPipe(final TCreatePipeReq req) {
124-
return new PipeParameters(req.getExtractorAttributes())
125-
.getBooleanOrDefault(
126-
Arrays.asList(
127-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
128-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
129-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
124+
return PipeSourceConstant.isDoubleLiving(new PipeParameters(req.getExtractorAttributes()));
130125
}
131126

132127
private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) {
133128
final PipeParameters sourceParameters = new PipeParameters(req.getExtractorAttributes());
129+
final TSStatus validationStatus = validateDoubleLivingPipeParameters(sourceParameters);
130+
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != validationStatus.getCode()) {
131+
return validationStatus;
132+
}
134133
final String currentDialect =
135134
sourceParameters.getStringOrDefault(
136135
SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE);
@@ -180,11 +179,24 @@ private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) {
180179
return secondStatus;
181180
}
182181

182+
private TSStatus validateDoubleLivingPipeParameters(final PipeParameters sourceParameters) {
183+
final Boolean isForwardingPipeRequests =
184+
sourceParameters.getBooleanByKeys(
185+
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
186+
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY);
187+
return Boolean.TRUE.equals(isForwardingPipeRequests)
188+
? RpcUtils.getStatus(
189+
TSStatusCode.PIPE_ERROR,
190+
PipeMessages
191+
.EXCEPTION_FORWARDING_PIPE_REQUESTS_CAN_NOT_SPECIFIED_TRUE_DOUBLE_LIVING_ENABLED_B000E8A1)
192+
: RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
193+
}
194+
183195
private TCreatePipeReq cloneCreatePipeRequestWithDialect(
184196
final TCreatePipeReq req, final PipeParameters sourceParameters, final String sqlDialect) {
185197
final Map<String, String> sourceAttributes = new HashMap<>(sourceParameters.getAttribute());
186-
sourceAttributes.remove(PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY);
187-
sourceAttributes.remove(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
198+
PipeSourceConstant.removeDoubleLivingAttributes(sourceAttributes);
199+
PipeSourceConstant.disableForwardingPipeRequests(sourceAttributes);
188200
sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
189201
sourceAttributes.put(
190202
SystemConstant.PIPE_VISIBILITY_KEY, SystemConstant.PIPE_VISIBILITY_STRICT_VALUE);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -395,20 +395,13 @@ public void customize(
395395

396396
skipIfNoPrivileges = getSkipIfNoPrivileges(parameters);
397397

398-
final boolean isDoubleLiving =
399-
parameters.getBooleanOrDefault(
400-
Arrays.asList(
401-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
402-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
403-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
398+
final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
404399
if (isDoubleLiving) {
405400
isForwardingPipeRequests = false;
406401
} else {
407402
isForwardingPipeRequests =
408403
parameters.getBooleanOrDefault(
409-
Arrays.asList(
410-
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
411-
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
404+
PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
412405
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
413406
}
414407

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -262,20 +262,13 @@ public void customize(
262262
? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime)
263263
: TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1;
264264

265-
final boolean isDoubleLiving =
266-
parameters.getBooleanOrDefault(
267-
Arrays.asList(
268-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
269-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
270-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
265+
final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
271266
if (isDoubleLiving) {
272267
isForwardingPipeRequests = false;
273268
} else {
274269
isForwardingPipeRequests =
275270
parameters.getBooleanOrDefault(
276-
Arrays.asList(
277-
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
278-
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
271+
PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
279272
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
280273
}
281274

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
import org.apache.iotdb.commons.exception.SemanticException;
6161
import org.apache.iotdb.commons.executable.ExecutableManager;
6262
import org.apache.iotdb.commons.executable.ExecutableResource;
63+
import org.apache.iotdb.commons.i18n.PipeMessages;
6364
import org.apache.iotdb.commons.path.MeasurementPath;
6465
import org.apache.iotdb.commons.path.PartialPath;
6566
import org.apache.iotdb.commons.path.PathPatternTree;
@@ -347,6 +348,7 @@
347348
import org.apache.iotdb.db.trigger.service.TriggerClassLoader;
348349
import org.apache.iotdb.pipe.api.PipePlugin;
349350
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
351+
import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
350352
import org.apache.iotdb.pipe.api.exception.PipePasswordCheckException;
351353
import org.apache.iotdb.rpc.RpcUtils;
352354
import org.apache.iotdb.rpc.StatementExecutionException;
@@ -2278,6 +2280,7 @@ public SettableFuture<ConfigTaskResult> createPipe(
22782280
// Validate pipe plugin before creation
22792281
try {
22802282
if (isDoubleLivingPipe(sourcePipeParameters)) {
2283+
validateDoubleLivingPipeParameters(sourcePipeParameters);
22812284
validatePipePlugin(
22822285
pipeName,
22832286
cloneSourceParametersWithDialect(
@@ -2342,18 +2345,26 @@ private void validatePipePlugin(
23422345
}
23432346

23442347
private boolean isDoubleLivingPipe(final PipeParameters sourcePipeParameters) {
2345-
return sourcePipeParameters.getBooleanOrDefault(
2346-
Arrays.asList(
2347-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
2348-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
2349-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
2348+
return PipeSourceConstant.isDoubleLiving(sourcePipeParameters);
2349+
}
2350+
2351+
private void validateDoubleLivingPipeParameters(final PipeParameters sourcePipeParameters) {
2352+
final Boolean isForwardingPipeRequests =
2353+
sourcePipeParameters.getBooleanByKeys(
2354+
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
2355+
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY);
2356+
if (Boolean.TRUE.equals(isForwardingPipeRequests)) {
2357+
throw new PipeParameterNotValidException(
2358+
PipeMessages
2359+
.EXCEPTION_FORWARDING_PIPE_REQUESTS_CAN_NOT_SPECIFIED_TRUE_DOUBLE_LIVING_ENABLED_B000E8A1);
2360+
}
23502361
}
23512362

23522363
private PipeParameters cloneSourceParametersWithDialect(
23532364
final PipeParameters sourcePipeParameters, final String sqlDialect) {
23542365
final Map<String, String> sourceAttributes = new HashMap<>(sourcePipeParameters.getAttribute());
2355-
sourceAttributes.remove(PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY);
2356-
sourceAttributes.remove(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
2366+
PipeSourceConstant.removeDoubleLivingAttributes(sourceAttributes);
2367+
PipeSourceConstant.disableForwardingPipeRequests(sourceAttributes);
23572368
sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
23582369
sourceAttributes.put(
23592370
SystemConstant.PIPE_VISIBILITY_KEY, SystemConstant.PIPE_VISIBILITY_STRICT_VALUE);

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,13 @@
2020
package org.apache.iotdb.commons.pipe.config.constant;
2121

2222
import org.apache.iotdb.commons.i18n.PipeMessages;
23+
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
2324

2425
import java.util.Arrays;
2526
import java.util.Collections;
2627
import java.util.HashSet;
28+
import java.util.List;
29+
import java.util.Map;
2730
import java.util.Set;
2831

2932
public class PipeSourceConstant {
@@ -87,6 +90,10 @@ public class PipeSourceConstant {
8790
public static final String SOURCE_FORWARDING_PIPE_REQUESTS_KEY =
8891
"source.forwarding-pipe-requests";
8992
public static final boolean EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE = true;
93+
public static final List<String> FORWARDING_PIPE_REQUESTS_KEYS =
94+
Collections.unmodifiableList(
95+
Arrays.asList(
96+
EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY, SOURCE_FORWARDING_PIPE_REQUESTS_KEY));
9097

9198
public static final String EXTRACTOR_HISTORY_ENABLE_KEY = "extractor.history.enable";
9299
public static final String SOURCE_HISTORY_ENABLE_KEY = "source.history.enable";
@@ -137,7 +144,16 @@ public class PipeSourceConstant {
137144
public static final boolean EXTRACTOR_MODE_SNAPSHOT_DEFAULT_VALUE = false;
138145
public static final String EXTRACTOR_MODE_DOUBLE_LIVING_KEY = "extractor.mode.double-living";
139146
public static final String SOURCE_MODE_DOUBLE_LIVING_KEY = "source.mode.double-living";
147+
public static final String EXTRACTOR_DOUBLE_LIVING_KEY = "extractor.double-living";
148+
public static final String SOURCE_DOUBLE_LIVING_KEY = "source.double-living";
140149
public static final boolean EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE = false;
150+
public static final List<String> DOUBLE_LIVING_KEYS =
151+
Collections.unmodifiableList(
152+
Arrays.asList(
153+
EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
154+
SOURCE_MODE_DOUBLE_LIVING_KEY,
155+
EXTRACTOR_DOUBLE_LIVING_KEY,
156+
SOURCE_DOUBLE_LIVING_KEY));
141157

142158
public static final String EXTRACTOR_START_TIME_KEY = "extractor.start-time";
143159
public static final String SOURCE_START_TIME_KEY = "source.start-time";
@@ -195,6 +211,31 @@ public class PipeSourceConstant {
195211
public static final String EXTRACTOR_CONSENSUS_RECEIVER_DATANODE_ID_KEY =
196212
"extractor.consensus.receiver-dn-id";
197213

214+
public static boolean isDoubleLiving(final PipeParameters sourceParameters) {
215+
return sourceParameters.getBooleanOrDefault(
216+
DOUBLE_LIVING_KEYS, EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
217+
}
218+
219+
public static void removeDoubleLivingAttributes(final Map<String, String> sourceAttributes) {
220+
removeEquivalentAttributes(sourceAttributes, DOUBLE_LIVING_KEYS);
221+
}
222+
223+
public static void disableForwardingPipeRequests(final Map<String, String> sourceAttributes) {
224+
removeEquivalentAttributes(sourceAttributes, FORWARDING_PIPE_REQUESTS_KEYS);
225+
sourceAttributes.put(SOURCE_FORWARDING_PIPE_REQUESTS_KEY, Boolean.FALSE.toString());
226+
}
227+
228+
private static void removeEquivalentAttributes(
229+
final Map<String, String> sourceAttributes, final List<String> equivalentKeys) {
230+
final Set<String> reducedKeys =
231+
equivalentKeys.stream()
232+
.map(PipeParameters.KeyReducer::reduce)
233+
.collect(java.util.stream.Collectors.toSet());
234+
sourceAttributes
235+
.keySet()
236+
.removeIf(key -> reducedKeys.contains(PipeParameters.KeyReducer.reduce(key)));
237+
}
238+
198239
private PipeSourceConstant() {
199240
throw new IllegalStateException(PipeMessages.UTILITY_CLASS);
200241
}

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,16 +27,10 @@
2727
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
2828

2929
import java.util.Arrays;
30-
import java.util.List;
3130
import java.util.Objects;
3231

3332
public class VisibilityUtils {
3433

35-
private static final List<String> DOUBLE_LIVING_KEYS =
36-
Arrays.asList(
37-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
38-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
39-
4034
private VisibilityUtils() {
4135
// forbidding instantiation
4236
}
@@ -91,8 +85,7 @@ public static Visibility calculateFromExtractorParameters(
9185
return calculateStrictlyFromExtractorParameters(extractorParameters);
9286
}
9387

94-
if (extractorParameters.getBooleanOrDefault(
95-
DOUBLE_LIVING_KEYS, PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE)) {
88+
if (PipeSourceConstant.isDoubleLiving(extractorParameters)) {
9689
return Visibility.BOTH;
9790
}
9891

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java‎

Lines changed: 3 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -115,12 +115,7 @@ public void validate(final PipeParameterValidator validator) throws Exception {
115115
}
116116

117117
private void validateDoubleLiving(final PipeParameters parameters) {
118-
final boolean isDoubleLiving =
119-
parameters.getBooleanOrDefault(
120-
Arrays.asList(
121-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
122-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
123-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
118+
final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
124119
if (!isDoubleLiving) {
125120
return;
126121
}
@@ -149,20 +144,13 @@ public void customize(
149144
pipeTaskMeta = ((PipeTaskSourceRuntimeEnvironment) environment).getPipeTaskMeta();
150145
}
151146

152-
final boolean isDoubleLiving =
153-
parameters.getBooleanOrDefault(
154-
Arrays.asList(
155-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
156-
PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
157-
PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
147+
final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
158148
if (isDoubleLiving) {
159149
isForwardingPipeRequests = false;
160150
} else {
161151
isForwardingPipeRequests =
162152
parameters.getBooleanOrDefault(
163-
Arrays.asList(
164-
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
165-
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
153+
PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
166154
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
167155
}
168156

‎iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,10 @@ public void testLegacyVisibilityKeepsDoubleLivingAndCaptureAttributes() {
7070
doubleLivingAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY, "false");
7171
assertVisibility(Visibility.BOTH, doubleLivingAttributes, true, true);
7272

73+
final Map<String, String> directDoubleLivingAttributes = new HashMap<>();
74+
directDoubleLivingAttributes.put("double-living", "true");
75+
assertVisibility(Visibility.BOTH, directDoubleLivingAttributes, true, true);
76+
7377
final Map<String, String> captureBothAttributes = new HashMap<>();
7478
captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY, "true");
7579
captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY, "true");
@@ -81,6 +85,26 @@ public void testLegacyVisibilityKeepsDoubleLivingAndCaptureAttributes() {
8185
assertVisibility(Visibility.NONE, captureNoneAttributes, false, false);
8286
}
8387

88+
@Test
89+
public void testDoubleLivingSugarCanBeNormalizedForStrictPipes() {
90+
final Map<String, String> attributes = new HashMap<>();
91+
attributes.put("double-living", "true");
92+
attributes.put("mode.double-living", "true");
93+
attributes.put(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY, "true");
94+
attributes.put("forwarding-pipe-requests", "true");
95+
attributes.put(PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY, "true");
96+
97+
PipeSourceConstant.removeDoubleLivingAttributes(attributes);
98+
PipeSourceConstant.disableForwardingPipeRequests(attributes);
99+
100+
final PipeParameters parameters = new PipeParameters(attributes);
101+
Assert.assertFalse(PipeSourceConstant.isDoubleLiving(parameters));
102+
Assert.assertFalse(
103+
parameters.getBooleanOrDefault(
104+
PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
105+
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE));
106+
}
107+
84108
private void assertVisibility(
85109
final Visibility expectedVisibility,
86110
final Map<String, String> attributes,

0 commit comments

Comments
 (0)