czy006 commented on code in PR #4064:
URL: https://github.com/apache/flink-cdc/pull/4064#discussion_r2250876177
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java:
##########
@@ -171,8 +173,33 @@ public Iterator<SourceRecords> pollSplitRecords() throws
InterruptedException {
Optional<SourceRecord> oscRecord =
parseOnLineSchemaChangeEvent(event.getRecord());
if (oscRecord.isPresent()) {
- sourceRecords.add(oscRecord.get());
- continue;
+ TableId tableId =
RecordUtils.getTableId(oscRecord.get());
+ if (tableId != null) {
+ LOG.info(
+ "Received the start event of online schema
change: {}. Save it for later.",
+ oscRecord.get());
+ pendingSchemaChangeEvents.put(tableId.toString(),
oscRecord.get());
+ continue;
+ }
+ }
+
+ Optional<String> finishedTables =
+
RecordUtils.parseOnLineSchemaRenameEvent(event.getRecord());
+ if (finishedTables.isPresent()) {
+ TableId tableId =
RecordUtils.getTableId(event.getRecord());
Review Comment:
```
2025-08-04 16:57:01,623 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Checking if
DDL might be an OSC renaming event... rename /* gh-ost */ table
`dinky`.`dinky_user` to `dinky`.`_dinky_user_del`
2025-08-04 16:57:01,624 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Determined
the shorter TableId dinky_user is the renaming source.
2025-08-04 16:57:01,624 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Determined
the longer TableId _dinky_user_del is the renaming target.
2025-08-04 16:57:01,624 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Renamed to
TableId name _dinky_user_del matches OSC temporary TableId pattern, yield
dinky_user.
2025-08-04 16:57:01,624 INFO
org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader -
Received the ending event of
tableId:dinky._dinky_user_del,dinky_user,finishedTableId:dinky.dinky_user
2025-08-04 16:57:01,624 INFO
org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader -
Received the ending event of table dinky.dinky_user. Emit corresponding DDL
event now.
2025-08-04 16:57:01,624 INFO
org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader -
remove pendingSchemaChangeEvents:dinky.dinky_user
2025-08-04 16:57:01,626 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Checking if
DDL might be an OSC renaming event... rename /* gh-ost */ table
`dinky`.`_dinky_user_gho` to `dinky`.`dinky_user`
2025-08-04 16:57:01,626 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Determined
the shorter TableId dinky_user is the renaming source.
2025-08-04 16:57:01,626 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Determined
the longer TableId _dinky_user_del is the renaming target.
2025-08-04 16:57:01,626 INFO
org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils - Renamed to
TableId name _dinky_user_del matches OSC temporary TableId pattern, yield
dinky_user.
2025-08-04 16:57:01,626 INFO
org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader -
Received the ending event of
tableId:dinky._dinky_user_del,dinky_user,finishedTableId:dinky.dinky_user
2025-08-04 16:57:01,626 INFO
org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader -
Received the ending event of table dinky.dinky_user. Emit corresponding DDL
event now.
2025-08-04 16:57:01,626 ERROR
org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader -
Error: met an unexpected osc finish event. Current pending events: {}, Record:
DataChangeEvent
[record=SourceRecord{sourcePartition={server=mysql_binlog_source},
sourceOffset={transaction_id=null, ts_sec=1754297821, file=mysql-bin.000003,
pos=181321, server_id=1}} ConnectRecord{topic='mysql_binlog_source',
kafkaPartition=0, key=Struct{databaseName=dinky},
value=Struct{source=Struct{version=1.9.8.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1754297821468,db=dinky,table=_dinky_user_del,dinky_user,server_id=1,file=mysql-bin.000003,pos=181119,row=0},historyRecord={"source":{"file":"mysql-bin.000003","pos":181119,"server_id":1},"position
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
