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

DBZ-7608 Maintain order for schema events

Browse Source
This commit is contained in:
mfvitale 2024-03-07 14:01:34 +01:00 committed by Jiri Pechanec
parent 5a45d8dcf4
commit b8dcb958cd
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlSnapshotChangeEventSource.java
View File

@ -15,6 +15,7 @@
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@ -68,7 +69,7 @@ public class MySqlSnapshotChangeEventSource extends RelationalSnapshotChangeEven
private final RelationalTableFilters filters;
private final MySqlSnapshotChangeEventSourceMetrics metrics;
private final MySqlDatabaseSchema databaseSchema;
private final Set<SchemaChangeEvent> schemaEvents = new HashSet<>();
private final Set<SchemaChangeEvent> schemaEvents = new LinkedHashSet<>();
private Set<TableId> delayedSchemaSnapshotTables = Collections.emptySet();
private final BlockingConsumer<Function<SourceRecord, SourceRecord>> lastEventProcessor;
private final Runnable preSnapshotAction;
@ -312,7 +313,7 @@ private void addSchemaEvent(RelationalSnapshotContext<MySqlPartition, MySqlOffse
List<SchemaChangeEvent> schemaChangeEvents = databaseSchema.parseSnapshotDdl(snapshotContext.partition, ddl, database,
snapshotContext.offset, clock.currentTimeAsInstant());
schemaEvents.addAll(new HashSet<>(schemaChangeEvents));
schemaEvents.addAll(new LinkedHashSet<>(schemaChangeEvents));
}
@Override