alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-5386 Missing "previousId" property with parsing the rename statement in kafka history topic

Browse Source
This commit is contained in:
harveyyue 2022-07-10 00:31:18 +08:00 committed by Jiri Pechanec
parent 366c95707a
commit 7ba7929d7e
1 changed files with 24 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlDatabaseSchema.java
View File

@ -291,15 +291,30 @@ else if (event instanceof SetVariableEvent) {
private void emitChangeEvent(MySqlPartition partition, MySqlOffsetContext offset, List<SchemaChangeEvent> schemaChangeEvents,
final String sanitizedDbName, Event event, TableId tableId, SchemaChangeEventType type,
boolean snapshot) {
schemaChangeEvents.add(SchemaChangeEvent.of(
type,
partition,
offset,
sanitizedDbName,
null,
event.statement(),
tableId != null ? tableFor(tableId) : null,
snapshot));
SchemaChangeEvent schemaChangeEvent;
if (type.equals(SchemaChangeEventType.ALTER) && event instanceof TableAlteredEvent
&& ((TableAlteredEvent) event).previousTableId() != null) {
schemaChangeEvent = SchemaChangeEvent.ofRename(
partition,
offset,
sanitizedDbName,
null,
event.statement(),
tableId != null ? tableFor(tableId) : null,
((TableAlteredEvent) event).previousTableId());
}
else {
schemaChangeEvent = SchemaChangeEvent.of(
type,
partition,
offset,
sanitizedDbName,
null,
event.statement(),
tableId != null ? tableFor(tableId) : null,
snapshot);
}
schemaChangeEvents.add(schemaChangeEvent);
}
private boolean acceptableDatabase(final String databaseName) {