Skip to content

Commit 45da56a

Browse files
CaideyipiCaideyipi
committed
Pipe: disable unused logic in open source
1 parent 1c20df4 commit 45da56a

44 files changed

Lines changed: 147 additions & 1367 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public void testBasicAlterPipe() throws Exception {
6363
// The database & table name will be converted to lower case
6464
final String sql =
6565
String.format(
66-
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='Test', 'table-name'='Test1', 'mode.streaming'='true') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
66+
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='Test', 'table-name'='Test1', 'source.realtime.mode'='batch') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
6767
receiverDataNode.getIpAndPortString());
6868
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
6969
final Statement statement = connection.createStatement()) {
@@ -88,7 +88,7 @@ public void testBasicAlterPipe() throws Exception {
8888
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
8989
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
9090
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
91-
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
91+
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=batch"));
9292
Assert.assertTrue(
9393
showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
9494
Assert.assertTrue(
@@ -146,7 +146,7 @@ public void testBasicAlterPipe() throws Exception {
146146
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
147147
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test1"));
148148
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test1"));
149-
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
149+
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=batch"));
150150
Assert.assertTrue(
151151
showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
152152
Assert.assertTrue(
@@ -185,7 +185,8 @@ public void testBasicAlterPipe() throws Exception {
185185
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
186186
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
187187
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
188-
Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
188+
Assert.assertFalse(
189+
showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=batch"));
189190
Assert.assertTrue(
190191
showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
191192
Assert.assertTrue(
@@ -443,7 +444,7 @@ public void testAlterPipeFailure() {
443444
// Create pipe
444445
sql =
445446
String.format(
446-
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test1', 'mode.streaming'='true') with sink ('node-urls'='%s', 'batch.enable'='false')",
447+
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test1', 'source.realtime.mode'='batch') with sink ('node-urls'='%s', 'batch.enable'='false')",
447448
receiverDataNode.getIpAndPortString());
448449
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
449450
final Statement statement = connection.createStatement()) {
@@ -468,7 +469,7 @@ public void testAlterPipeSourceAndSink() {
468469
// Create pipe
469470
final String sql =
470471
String.format(
471-
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test', 'mode.streaming'='true') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s', 'batch.enable'='false')",
472+
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test', 'source.realtime.mode'='batch') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s', 'batch.enable'='false')",
472473
receiverDataNode.getIpAndPortString());
473474
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
474475
final Statement statement = connection.createStatement()) {
@@ -488,7 +489,7 @@ public void testAlterPipeSourceAndSink() {
488489
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
489490
final Statement statement = connection.createStatement()) {
490491
statement.execute(
491-
"alter pipe a2b modify source('source' = 'iotdb-source','database-name'='test1', 'table-name'='test1', 'mode.streaming'='true', 'source.inclusion'='data.insert') modify sink ('batch.enable'='true')");
492+
"alter pipe a2b modify source('source' = 'iotdb-source','database-name'='test1', 'table-name'='test1', 'source.realtime.mode'='batch', 'source.inclusion'='data.insert') modify sink ('batch.enable'='true')");
492493
} catch (final SQLException e) {
493494
fail(e.getMessage());
494495
}

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {
7777

7878
TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
7979
TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
80+
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
8081

8182
TestUtils.executeNonQueries(
8283
senderEnv,
@@ -87,7 +88,7 @@ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {
8788
final Map<String, String> processorAttributes = new HashMap<>();
8889
final Map<String, String> sinkAttributes = new HashMap<>();
8990

90-
sourceAttributes.put("source.realtime.mode", "log");
91+
sourceAttributes.put("source.realtime.mode", "batch");
9192
sourceAttributes.put("capture.table", "true");
9293
sourceAttributes.put("__system.sql-dialect", "table");
9394
sourceAttributes.put("capture.tree", "true");
@@ -112,6 +113,7 @@ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {
112113
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
113114

114115
TableModelUtils.insertData("test", "test", 50, 100, senderEnv, true);
116+
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
115117

116118
TestUtils.executeNonQueries(
117119
senderEnv,
@@ -144,11 +146,6 @@ public void testTsFileFormatAndAsyncLoad() throws Exception {
144146
testSinkFormat("tsfile", true);
145147
}
146148

147-
@Test
148-
public void testSinkHybridFormat() throws Exception {
149-
testSinkFormat("hybrid", false);
150-
}
151-
152149
private void testSinkFormat(final String format, final boolean isAsyncLoad) throws Exception {
153150
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
154151

@@ -165,6 +162,7 @@ private void testSinkFormat(final String format, final boolean isAsyncLoad) thro
165162

166163
TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
167164
TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
165+
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
168166

169167
TestUtils.executeNonQueries(
170168
senderEnv,
@@ -202,6 +200,7 @@ private void testSinkFormat(final String format, final boolean isAsyncLoad) thro
202200
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
203201

204202
TableModelUtils.insertData("test", "test", 50, 150, senderEnv, true);
203+
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
205204

206205
TestUtils.executeNonQueries(
207206
senderEnv,
@@ -244,6 +243,7 @@ private void testSinkFormat(final String format, final boolean isAsyncLoad) thro
244243
null);
245244

246245
TableModelUtils.insertData("test", "test", 150, 200, senderEnv, true);
246+
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
247247
TableModelUtils.insertTablet("test", "test", 200, 250, senderEnv, true);
248248
TableModelUtils.insertTablet("test", "test", 250, 300, senderEnv, true);
249249
TableModelUtils.insertTablet("test", "test", 300, 350, senderEnv, true);

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

Lines changed: 2 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -204,7 +204,7 @@ public void testLifeCycleLogMode() throws Exception {
204204

205205
sourceAttributes.put("capture.table", "true");
206206
sourceAttributes.put("__system.sql-dialect", "table");
207-
sourceAttributes.put("source.mode", "forced-log");
207+
sourceAttributes.put("source.realtime.mode", "batch");
208208
sourceAttributes.put("user", "root");
209209

210210
sinkAttributes.put("sink", "iotdb-thrift-sink");
@@ -261,7 +261,7 @@ public void testLifeCycleFileMode() throws Exception {
261261

262262
sourceAttributes.put("capture.table", "true");
263263
sourceAttributes.put("__system.sql-dialect", "table");
264-
sourceAttributes.put("mode.streaming", "false");
264+
sourceAttributes.put("source.realtime.mode", "batch");
265265
sourceAttributes.put("user", "root");
266266

267267
sinkAttributes.put("sink", "iotdb-thrift-sink");
@@ -292,59 +292,6 @@ public void testLifeCycleFileMode() throws Exception {
292292
}
293293
}
294294

295-
@Test
296-
public void testLifeCycleHybridMode() throws Exception {
297-
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
298-
299-
final String receiverIp = receiverDataNode.getIp();
300-
final int receiverPort = receiverDataNode.getPort();
301-
final Consumer<String> handleFailure =
302-
o -> {
303-
executeNonQueryWithRetry(senderEnv, "flush");
304-
executeNonQueryWithRetry(receiverEnv, "flush");
305-
};
306-
307-
try (final SyncConfigNodeIServiceClient client =
308-
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
309-
310-
TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
311-
TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
312-
313-
final Map<String, String> sourceAttributes = new HashMap<>();
314-
final Map<String, String> processorAttributes = new HashMap<>();
315-
final Map<String, String> sinkAttributes = new HashMap<>();
316-
317-
sourceAttributes.put("capture.table", "true");
318-
sourceAttributes.put("__system.sql-dialect", "table");
319-
sourceAttributes.put("source.mode", "hybrid");
320-
sourceAttributes.put("user", "root");
321-
322-
sinkAttributes.put("sink", "iotdb-thrift-sink");
323-
sinkAttributes.put("sink.batch.enable", "false");
324-
sinkAttributes.put("sink.ip", receiverIp);
325-
sinkAttributes.put("sink.port", Integer.toString(receiverPort));
326-
327-
final TSStatus status =
328-
client.createPipe(
329-
new TCreatePipeReq("p1", sinkAttributes)
330-
.setExtractorAttributes(sourceAttributes)
331-
.setProcessorAttributes(processorAttributes));
332-
333-
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
334-
335-
Assert.assertEquals(
336-
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
337-
338-
TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
339-
340-
TableModelUtils.assertCountData("test", "test", 200, receiverEnv, handleFailure);
341-
342-
TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
343-
344-
TableModelUtils.assertCountData("test", "test", 300, receiverEnv, handleFailure);
345-
}
346-
}
347-
348295
@Test
349296
public void testLifeCycleWithClusterRestart() throws Exception {
350297
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

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

Lines changed: 7 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -219,8 +219,7 @@ public void testPipeOnBothSenderAndReceiver() throws Exception {
219219
extractorAttributes.put("database-name", "test");
220220
extractorAttributes.put("table-name", "test.*");
221221
extractorAttributes.put("inclusion", "data.insert");
222-
extractorAttributes.put("mode.streaming", "true");
223-
extractorAttributes.put("mode.snapshot", "false");
222+
extractorAttributes.put("source.realtime.mode", "batch");
224223
extractorAttributes.put("mode.strict", "true");
225224
extractorAttributes.put("user", "root");
226225

@@ -263,8 +262,7 @@ public void testPipeOnBothSenderAndReceiver() throws Exception {
263262
extractorAttributes.put("database-name", "test.*");
264263
extractorAttributes.put("table-name", "test.*");
265264
extractorAttributes.put("inclusion", "data.insert");
266-
extractorAttributes.put("mode.streaming", "true");
267-
extractorAttributes.put("mode.snapshot", "false");
265+
extractorAttributes.put("source.realtime.mode", "batch");
268266
extractorAttributes.put("mode.strict", "true");
269267
extractorAttributes.put("user", "root");
270268

@@ -313,8 +311,7 @@ private void doTest() throws Exception {
313311
extractorAttributes.put("database-name", "test.*");
314312
extractorAttributes.put("table-name", "test.*");
315313
extractorAttributes.put("inclusion", "data.insert");
316-
extractorAttributes.put("mode.streaming", "true");
317-
extractorAttributes.put("mode.snapshot", "false");
314+
extractorAttributes.put("source.realtime.mode", "batch");
318315
extractorAttributes.put("mode.strict", "true");
319316
extractorAttributes.put("user", "root");
320317

@@ -358,11 +355,6 @@ public void testAsyncConnectorUseNodeUrls() throws Exception {
358355
doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName());
359356
}
360357

361-
@Test
362-
public void testAirGapConnectorUseNodeUrls() throws Exception {
363-
doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName());
364-
}
365-
366358
private void doTestUseNodeUrls(String connectorName) throws Exception {
367359
senderEnv
368360
.getConfig()
@@ -407,16 +399,7 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {
407399
};
408400

409401
for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) {
410-
if (connectorName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) {
411-
// Use default port for convenience
412-
nodeUrlsBuilder
413-
.append(wrapper.getIp())
414-
.append(":")
415-
.append(wrapper.getPipeAirGapReceiverPort())
416-
.append(",");
417-
} else {
418-
nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
419-
}
402+
nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
420403
}
421404

422405
try (final SyncConfigNodeIServiceClient client =
@@ -440,12 +423,11 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {
440423
extractorAttributes.put("database-name", "test.*");
441424
extractorAttributes.put("table-name", "test.*");
442425
extractorAttributes.put("inclusion", "data.insert");
443-
extractorAttributes.put("mode.snapshot", "false");
444426
extractorAttributes.put("mode.strict", "true");
445427
extractorAttributes.put("user", "root");
446428

447-
// Test forced-log mode, in open releases this might be "file"
448-
extractorAttributes.put("realtime.mode", "forced-log");
429+
// Test batch mode
430+
extractorAttributes.put("realtime.mode", "batch");
449431

450432
TSStatus status =
451433
client.createPipe(
@@ -465,7 +447,7 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {
465447

466448
TableModelUtils.assertData("test", "test", 0, 300, receiverEnv, handleFailure);
467449

468-
extractorAttributes.replace("realtime.mode", "file");
450+
extractorAttributes.replace("realtime.mode", "batch");
469451

470452
status =
471453
client.createPipe(

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

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -71,8 +71,7 @@ public void testPipeSwitchStatus() throws Exception {
7171
extractorAttributes.put("database-name", "test");
7272
extractorAttributes.put("table-name", "test.*");
7373
extractorAttributes.put("inclusion", "data.insert");
74-
extractorAttributes.put("mode.streaming", "true");
75-
extractorAttributes.put("mode.snapshot", "false");
74+
extractorAttributes.put("source.realtime.mode", "batch");
7675
extractorAttributes.put("mode.strict", "true");
7776
extractorAttributes.put("start-time", "1");
7877
extractorAttributes.put("end-time", "2");
@@ -170,8 +169,7 @@ public void testPipeIllegallySwitchStatus() throws Exception {
170169
extractorAttributes.put("database-name", "test");
171170
extractorAttributes.put("table-name", "test.*");
172171
extractorAttributes.put("inclusion", "data.insert");
173-
extractorAttributes.put("mode.streaming", "true");
174-
extractorAttributes.put("mode.snapshot", "false");
172+
extractorAttributes.put("source.realtime.mode", "batch");
175173
extractorAttributes.put("mode.strict", "true");
176174
extractorAttributes.put("start-time", "1");
177175
extractorAttributes.put("end-time", "2");
@@ -257,8 +255,7 @@ public void testDropPipeAndCreateAgain() throws Exception {
257255
extractorAttributes.put("database-name", "test");
258256
extractorAttributes.put("table-name", "test.*");
259257
extractorAttributes.put("inclusion", "data.insert");
260-
extractorAttributes.put("mode.streaming", "true");
261-
extractorAttributes.put("mode.snapshot", "false");
258+
extractorAttributes.put("source.realtime.mode", "batch");
262259
extractorAttributes.put("mode.strict", "true");
263260
extractorAttributes.put("start-time", "0");
264261
extractorAttributes.put("end-time", "200");
@@ -325,8 +322,7 @@ public void testWrongPipeName() throws Exception {
325322
extractorAttributes.put("database-name", "test");
326323
extractorAttributes.put("table-name", "test.*");
327324
extractorAttributes.put("inclusion", "data.insert");
328-
extractorAttributes.put("mode.streaming", "true");
329-
extractorAttributes.put("mode.snapshot", "false");
325+
extractorAttributes.put("source.realtime.mode", "batch");
330326
extractorAttributes.put("mode.strict", "true");
331327
extractorAttributes.put("start-time", "0");
332328
extractorAttributes.put("end-time", "200");

0 commit comments

Comments
 (0)