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

DBZ-7608 Resolve performance issue during addSchemaEvent

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

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

@ -5,8 +5,6 @@
*/
package io.debezium.connector.mysql;
import static java.util.function.Predicate.not;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@ -70,7 +68,7 @@ public class MySqlSnapshotChangeEventSource extends RelationalSnapshotChangeEven
private final RelationalTableFilters filters;
private final MySqlSnapshotChangeEventSourceMetrics metrics;
private final MySqlDatabaseSchema databaseSchema;
private final List<SchemaChangeEvent> schemaEvents = new ArrayList<>();
private final Set<SchemaChangeEvent> schemaEvents = new HashSet<>();
private Set<TableId> delayedSchemaSnapshotTables = Collections.emptySet();
private final BlockingConsumer<Function<SourceRecord, SourceRecord>> lastEventProcessor;
private final Runnable preSnapshotAction;
@ -314,10 +312,7 @@ private void addSchemaEvent(RelationalSnapshotContext<MySqlPartition, MySqlOffse
List<SchemaChangeEvent> schemaChangeEvents = databaseSchema.parseSnapshotDdl(snapshotContext.partition, ddl, database,
snapshotContext.offset, clock.currentTimeAsInstant());
List<SchemaChangeEvent> missingSchemaChangeEvents = schemaChangeEvents.stream()
.filter(not(schemaEvents::contains))
.collect(Collectors.toList());
schemaEvents.addAll(missingSchemaChangeEvents);
schemaEvents.addAll(new HashSet<>(schemaChangeEvents));
}
@Override