lvyanquan commented on code in PR #4069:
URL: https://github.com/apache/flink-cdc/pull/4069#discussion_r2250598523
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java:
##########
@@ -146,8 +148,11 @@ protected boolean isSchemaChangeRecord(SourceRecord
record) {
@Override
protected TableId getTableId(SourceRecord record) {
- String[] parts = record.topic().split("\\.");
- return TableId.tableId(parts[1], parts[2]);
+ Struct value = (Struct) record.value();
+ Struct source = value.getStruct(Envelope.FieldName.SOURCE);
+ String dbName = source.getString(DATABASE_NAME_KEY);
+ String tableName = source.getString(TABLE_NAME_KEY);
+ return TableId.tableId(dbName, tableName);
Review Comment:
Copy from
[SourceRecordUtils#getTableId](https://github.com/apache/flink-cdc/blob/5eaf9eb86b81a1fe80a06abe7c39a85cadc9b35e/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SourceRecordUtils.java#L119).
There may be issues when the table name contains special characters in
record.topic() method, for example, `sqlInjection; DROP TABLE important_data;`
will be converted to `sqlInjection__DROP_TABLE_important_data_`.
--
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
