|
60 | 60 | import org.junit.jupiter.api.BeforeAll; |
61 | 61 | import org.junit.jupiter.api.BeforeEach; |
62 | 62 | import org.junit.jupiter.api.Test; |
| 63 | +import org.junit.jupiter.params.ParameterizedTest; |
| 64 | +import org.junit.jupiter.params.provider.ValueSource; |
63 | 65 | import org.testcontainers.lifecycle.Startables; |
64 | 66 |
|
65 | 67 | import java.sql.Connection; |
@@ -399,7 +401,131 @@ void testSqlInjection() throws Exception { |
399 | 401 | } |
400 | 402 |
|
401 | 403 | @Test |
402 | | - void testExcludeTables() throws Exception { |
| 404 | + void testLatestOffsetStartupMode() throws Exception { |
| 405 | + inventoryDatabase.createAndInitialize(); |
| 406 | + MySqlSourceConfigFactory configFactory = |
| 407 | + new MySqlSourceConfigFactory() |
| 408 | + .hostname(MYSQL8_CONTAINER.getHost()) |
| 409 | + .port(MYSQL8_CONTAINER.getDatabasePort()) |
| 410 | + .username(TEST_USER) |
| 411 | + .password(TEST_PASSWORD) |
| 412 | + .databaseList(inventoryDatabase.getDatabaseName()) |
| 413 | + .tableList(inventoryDatabase.getDatabaseName() + "\\.products") |
| 414 | + .startupOptions(StartupOptions.latest()) |
| 415 | + .serverId(getServerId(env.getParallelism())) |
| 416 | + .serverTimeZone("UTC") |
| 417 | + .includeSchemaChanges(SCHEMA_CHANGE_ENABLED.defaultValue()); |
| 418 | + |
| 419 | + FlinkSourceProvider sourceProvider = |
| 420 | + (FlinkSourceProvider) new MySqlDataSource(configFactory).getEventSourceProvider(); |
| 421 | + CloseableIterator<Event> events = |
| 422 | + env.fromSource( |
| 423 | + sourceProvider.getSource(), |
| 424 | + WatermarkStrategy.noWatermarks(), |
| 425 | + MySqlDataSourceFactory.IDENTIFIER, |
| 426 | + new EventTypeInfo()) |
| 427 | + .executeAndCollect(); |
| 428 | + Thread.sleep(10_000); |
| 429 | + TableId tableId = TableId.tableId(inventoryDatabase.getDatabaseName(), "products"); |
| 430 | + |
| 431 | + List<Event> expectedBinlog = new ArrayList<>(); |
| 432 | + try (Connection connection = inventoryDatabase.getJdbcConnection(); |
| 433 | + Statement statement = connection.createStatement()) { |
| 434 | + expectedBinlog.addAll(executeAlterAndProvideExpected(tableId, statement)); |
| 435 | + |
| 436 | + RowType rowType = |
| 437 | + RowType.of( |
| 438 | + new DataType[] { |
| 439 | + DataTypes.INT().notNull(), |
| 440 | + DataTypes.VARCHAR(255).notNull(), |
| 441 | + DataTypes.FLOAT(), |
| 442 | + DataTypes.VARCHAR(45), |
| 443 | + DataTypes.VARCHAR(55) |
| 444 | + }, |
| 445 | + new String[] {"id", "name", "weight", "col1", "col2"}); |
| 446 | + BinaryRecordDataGenerator generator = new BinaryRecordDataGenerator(rowType); |
| 447 | + // insert more data |
| 448 | + statement.execute( |
| 449 | + String.format( |
| 450 | + "INSERT INTO `%s`.`products` VALUES (default,'scooter',5.5,'c-10','c-20');", |
| 451 | + inventoryDatabase.getDatabaseName())); // 110 |
| 452 | + expectedBinlog.add( |
| 453 | + DataChangeEvent.insertEvent( |
| 454 | + tableId, |
| 455 | + generator.generate( |
| 456 | + new Object[] { |
| 457 | + 110, |
| 458 | + BinaryStringData.fromString("scooter"), |
| 459 | + 5.5f, |
| 460 | + BinaryStringData.fromString("c-10"), |
| 461 | + BinaryStringData.fromString("c-20") |
| 462 | + }))); |
| 463 | + statement.execute( |
| 464 | + String.format( |
| 465 | + "INSERT INTO `%s`.`products` VALUES (default,'football',6.6,'c-11','c-21');", |
| 466 | + inventoryDatabase.getDatabaseName())); // 111 |
| 467 | + expectedBinlog.add( |
| 468 | + DataChangeEvent.insertEvent( |
| 469 | + tableId, |
| 470 | + generator.generate( |
| 471 | + new Object[] { |
| 472 | + 111, |
| 473 | + BinaryStringData.fromString("football"), |
| 474 | + 6.6f, |
| 475 | + BinaryStringData.fromString("c-11"), |
| 476 | + BinaryStringData.fromString("c-21") |
| 477 | + }))); |
| 478 | + statement.execute( |
| 479 | + String.format( |
| 480 | + "UPDATE `%s`.`products` SET `col1`='c-12', `col2`='c-22' WHERE id=110;", |
| 481 | + inventoryDatabase.getDatabaseName())); |
| 482 | + expectedBinlog.add( |
| 483 | + DataChangeEvent.updateEvent( |
| 484 | + tableId, |
| 485 | + generator.generate( |
| 486 | + new Object[] { |
| 487 | + 110, |
| 488 | + BinaryStringData.fromString("scooter"), |
| 489 | + 5.5f, |
| 490 | + BinaryStringData.fromString("c-10"), |
| 491 | + BinaryStringData.fromString("c-20") |
| 492 | + }), |
| 493 | + generator.generate( |
| 494 | + new Object[] { |
| 495 | + 110, |
| 496 | + BinaryStringData.fromString("scooter"), |
| 497 | + 5.5f, |
| 498 | + BinaryStringData.fromString("c-12"), |
| 499 | + BinaryStringData.fromString("c-22") |
| 500 | + }))); |
| 501 | + statement.execute( |
| 502 | + String.format( |
| 503 | + "DELETE FROM `%s`.`products` WHERE `id` = 111;", |
| 504 | + inventoryDatabase.getDatabaseName())); |
| 505 | + expectedBinlog.add( |
| 506 | + DataChangeEvent.deleteEvent( |
| 507 | + tableId, |
| 508 | + generator.generate( |
| 509 | + new Object[] { |
| 510 | + 111, |
| 511 | + BinaryStringData.fromString("football"), |
| 512 | + 6.6f, |
| 513 | + BinaryStringData.fromString("c-11"), |
| 514 | + BinaryStringData.fromString("c-21") |
| 515 | + }))); |
| 516 | + } |
| 517 | + // In this configuration, several subtasks might emit their corresponding CreateTableEvent |
| 518 | + // to downstream. Since it is not possible to predict how many CreateTableEvents should we |
| 519 | + // expect, we simply filter them out from expected sets, and assert there's at least one. |
| 520 | + |
| 521 | + Event createTableEvent = getProductsCreateTableEvent(tableId); |
| 522 | + List<Event> actual = fetchResultsExcept(events, expectedBinlog.size(), createTableEvent); |
| 523 | + assertThat(actual).isEqualTo(expectedBinlog); |
| 524 | + } |
| 525 | + |
| 526 | + @ParameterizedTest(name = "batchEmit: {0}") |
| 527 | + @ValueSource(booleans = {true, false}) |
| 528 | + void testExcludeTables(boolean inBatch) throws Exception { |
403 | 529 | inventoryDatabase.createAndInitialize(); |
404 | 530 | String databaseName = inventoryDatabase.getDatabaseName(); |
405 | 531 | MySqlSourceConfigFactory configFactory = |
|
0 commit comments