You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Qingsheng Ren (Jira)" <ji...@apache.org> on 2024/04/24 04:17:00 UTC
[jira] [Assigned] (FLINK-34869) [Bug][mysql] Remove all previous table and add new added table will throw Exception.
[ https://issues.apache.org/jira/browse/FLINK-34869?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Qingsheng Ren reassigned FLINK-34869:
-------------------------------------
Assignee: Josh Mahonin
> [Bug][mysql] Remove all previous table and add new added table will throw Exception.
> ------------------------------------------------------------------------------------
>
> Key: FLINK-34869
> URL: https://issues.apache.org/jira/browse/FLINK-34869
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Reporter: Flink CDC Issue Import
> Assignee: Josh Mahonin
> Priority: Major
> Labels: github-import
>
> ### Search before asking
> - [X] I searched in the [issues|https://github.com/ververica/flink-cdc-connectors/issues) and found nothing similar.
> ### Flink version
> 1.18
> ### Flink CDC version
> 3.0.1
> ### Database and its version
> anyone
> ### Minimal reproduce step
> 1. Stop job in savepoint.
> 2. Set 'scan.incremental.snapshot.enabled' = 'true' and then set tableList with tables which not includes in last time.
> 3. Then assign status will be chaos.
> Take a test case for example:
> ```java
> public class NewlyAddedTableITCase extends MySqlSourceTestBase {
> @Test
> public void testRemoveAndAddTablesOneByOne() throws Exception {
> testRemoveAndAddTablesOneByOne(
> 1, "address_hangzhou", "address_beijing", "address_shanghai");
> }
> private void testRemoveAndAddTablesOneByOne(int parallelism, String... captureAddressTables)
> throws Exception {
> MySqlConnection connection = getConnection();
> // step 1: create mysql tables with all tables included
> initialAddressTables(connection, captureAddressTables);
> final TemporaryFolder temporaryFolder = new TemporaryFolder();
> temporaryFolder.create();
> final String savepointDirectory = temporaryFolder.newFolder().toURI().toString();
> // get all expected data
> List<String> fetchedDataList = new ArrayList<>();
> String finishedSavePointPath = null;
> // test removing and adding table one by one
> for (int round = 0; round < captureAddressTables.length; round++] {
> String captureTableThisRound = captureAddressTables[round];
> String cityName = captureTableThisRound.split("_")[1];
> StreamExecutionEnvironment env =
> getStreamExecutionEnvironment(finishedSavePointPath, parallelism);
> StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
> String createTableStatement =
> getCreateTableStatement(new HashMap<>(), captureTableThisRound);
> tEnv.executeSql(createTableStatement);
> tEnv.executeSql(
> "CREATE TABLE sink ("
> + " table_name STRING,"
> + " id BIGINT,"
> + " country STRING,"
> + " city STRING,"
> + " detail_address STRING,"
> + " primary key (table_name,id) not enforced"
> + ") WITH ("
> + " 'connector' = 'values',"
> + " 'sink-insert-only' = 'false'"
> + ")");
> TableResult tableResult = tEnv.executeSql("insert into sink select * from address");
> JobClient jobClient = tableResult.getJobClient().get();
> // this round's snapshot data
> fetchedDataList.addAll(
> Arrays.asList(
> format(
> "+I[%s, 416874195632735147, China, %s, %s West Town address 1]",
> captureTableThisRound, cityName, cityName),
> format(
> "+I[%s, 416927583791428523, China, %s, %s West Town address 2]",
> captureTableThisRound, cityName, cityName),
> format(
> "+I[%s, 417022095255614379, China, %s, %s West Town address 3]",
> captureTableThisRound, cityName, cityName)));
> waitForSinkSize("sink", fetchedDataList.size());
> assertEqualsInAnyOrder(fetchedDataList, TestValuesTableFactory.getRawResults("sink"));
> // only this round table's data is captured.
> // step 3: make binlog data for all tables before this round(also includes this round)
> for (int i = 0; i <= round; i++) {
> String tableName = captureAddressTables[i];
> makeBinlogForAddressTable(connection, tableName, round);
> }
> // this round's binlog data
> fetchedDataList.addAll(
> Arrays.asList(
> format(
> "-U[%s, 416874195632735147, China, %s, %s West Town address 1]",
> captureTableThisRound, cityName, cityName),
> format(
> "+U[%s, 416874195632735147, CHINA_%s, %s, %s West Town address 1]",
> captureTableThisRound, round, cityName, cityName),
> format(
> "+I[%s, %d, China, %s, %s West Town address 4]",
> captureTableThisRound,
> 417022095255614380L + round,
> cityName,
> cityName)));
> // step 4: assert fetched binlog data in this round
> waitForSinkSize("sink", fetchedDataList.size());
> assertEqualsInAnyOrder(fetchedDataList, TestValuesTableFactory.getRawResults("sink"));
> // step 5: trigger savepoint
> finishedSavePointPath = triggerSavepointWithRetry(jobClient, savepointDirectory);
> jobClient.cancel().get();
> }
> }
> // setting primary key as id rather than <id, city> is more more realistic.
> private String getCreateTableStatement(
> Map<String, String> otherOptions, String... captureTableNames) {
> return format(
> "CREATE TABLE address ("
> + " table_name STRING METADATA VIRTUAL,"
> + " id BIGINT NOT NULL,"
> + " country STRING,"
> + " city STRING,"
> + " detail_address STRING,"
> + " primary key (id) not enforced"
> + ") WITH ("
> + " 'connector' = 'mysql-cdc',"
> + " 'scan.incremental.snapshot.enabled' = 'true',"
> + " 'hostname' = '%s',"
> + " 'port' = '%s',"
> + " 'username' = '%s',"
> + " 'password' = '%s',"
> + " 'database-name' = '%s',"
> + " 'table-name' = '%s',"
> + " 'scan.incremental.snapshot.chunk.size' = '2',"
> + " 'server-time-zone' = 'UTC',"
> + " 'server-id' = '%s',"
> + " 'scan.newly-added-table.enabled' = 'true'"
> + " %s"
> + ")",
> MYSQL_CONTAINER.getHost(),
> MYSQL_CONTAINER.getDatabasePort(),
> customDatabase.getUsername(),
> customDatabase.getPassword(),
> customDatabase.getDatabaseName(),
> getTableNameRegex(captureTableNames),
> getServerId(),
> otherOptions.isEmpty()
> ? ""
> : ","
> + otherOptions.entrySet().stream()
> .map(
> e ->
> String.format(
> "'%s'='%s'",
> e.getKey(), e.getValue()))
> .collect(Collectors.joining(",")));
> }
> }
> ```
> ### What did you expect to see?
> return true
> ### What did you see instead?
> An exception will occurs:
> ```java
> org.apache.flink.util.FlinkRuntimeException: The assigner is not ready to offer finished split information, this should not be called
> at com.ververica.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.getFinishedSplitInfos(MySqlSnapshotSplitAssigner.java:379) ~[classes/:?]
> at com.ververica.cdc.connectors.mysql.source.assigners.MySqlHybridSplitAssigner.getFinishedSplitInfos(MySqlHybridSplitAssigner.java:139) ~[classes/:?]
> at com.ververica.cdc.connectors.mysql.source.enumerator.MySqlSourceEnumerator.handleLatestFinishedSplitNumberRequest(MySqlSourceEnumerator.java:324) ~[classes/:?]
> at com.ververica.cdc.connectors.mysql.source.enumerator.MySqlSourceEnumerator.handleSourceEvent(MySqlSourceEnumerator.java:170) ~[classes/:?]
> at org.apache.flink.runtime.source.coordinator.SourceCoordinator.handleSourceEvent(SourceCoordinator.java:590) ~[flink-runtime-1.18.0.jar:1.18.0]
> at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$handleEventFromOperator$3(SourceCoordinator.java:297) ~[flink-runtime-1.18.0.jar:1.18.0]
> at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$10(SourceCoordinator.java:469) ~[flink-runtime-1.18.0.jar:1.18.0]
> at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) [flink-core-1.18.0.jar:1.18.0]
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_362]
> at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_362]
> at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_362]
> at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_362]
> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_362]
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_362]
> at java.lang.Thread.run(Thread.java:750) [?:1.8.0_362]
> ```
> ### Reason
> When restarted with newly added table, MySqlBinlogSplit#filterOutdatedSplitInfos will filter previous table' FinishedSnapshotSplitInfo. In this case, list of FinishedSnapshotSplitInfo will be empty.
> Then when add binlog split back to split reader, com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader#configureFilter cBinlogSplitReader#shouldEmit will seen an empty list of FinishedSnapshotSplitInfo as binlog-only job
> ```java
> // specific offset mode
> if (finishedSplitInfos.isEmpty()) {
> for (TableId tableId : currentBinlogSplit.getTableSchemas().keySet()) {
> tableIdBinlogPositionMap.put(tableId, currentBinlogSplit.getStartingOffset());
> }
> }
> // initial mode
> else {
> for (FinishedSnapshotSplitInfo finishedSplitInfo : finishedSplitInfos) {
> TableId tableId = finishedSplitInfo.getTableId();
> List<FinishedSnapshotSplitInfo> list =
> splitsInfoMap.getOrDefault(tableId, new ArrayList<>());
> list.add(finishedSplitInfo);
> splitsInfoMap.put(tableId, list);
> BinlogOffset highWatermark = finishedSplitInfo.getHighWatermark();
> BinlogOffset maxHighWatermark = tableIdBinlogPositionMap.get(tableId);
> if (maxHighWatermark == null || highWatermark.isAfter(maxHighWatermark)) {
> tableIdBinlogPositionMap.put(tableId, highWatermark);
> }
> }
> }
> ```
> ### Are you willing to submit a PR?
> - [x] I'm willing to submit a PR!
> ---------------- Imported from GitHub ----------------
> Url: https://github.com/apache/flink-cdc/issues/3051
> Created by: [loserwang1024|https://github.com/loserwang1024]
> Labels: bug,
> Created at: Wed Jan 31 17:12:21 CST 2024
> State: open
--
This message was sent by Atlassian Jira
(v8.20.10#820010)