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

DBZ-1482 Fix for mismatch source query during incremental snapshots

Browse Source
This commit is contained in:
Chris Cranford 2023-11-17 15:53:58 -05:00 committed by Jiri Pechanec
parent 3c3bd504c8
commit 1ce4ebd668
3 changed files with 21 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlOffsetContext.java
View File

@ -62,7 +62,6 @@ public MySqlOffsetContext(boolean snapshot, boolean snapshotCompleted, Transacti
public MySqlOffsetContext(MySqlConnectorConfig connectorConfig, boolean snapshot, boolean snapshotCompleted, SourceInfo sourceInfo) {
this(snapshot, snapshotCompleted, new TransactionContext(),
connectorConfig.getConnectorAdapter().getIncrementalSnapshotContext(),
// connectorConfig.isReadOnlyConnection() ? new MySqlReadOnlyIncrementalSnapshotContext<>() : new SignalBasedIncrementalSnapshotContext<>(),
sourceInfo);
}

22
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/strategy/mariadb/MariaDbConnectorAdapter.java
View File

@ -35,6 +35,7 @@
import io.debezium.relational.TableId;
import io.debezium.spi.schema.DataCollectionId;
import io.debezium.util.Clock;
import io.debezium.util.Strings;
/**
* This connector adapter provides a complete implementation for MariaDB assuming that
@ -100,7 +101,26 @@ else if (!connectorConfig.getSnapshotMode().shouldStream()) {
@Override
public String getRecordingQueryFromEvent(EventData eventData) {
return ((AnnotateRowsEventData) eventData).getRowsQuery();
final String query = ((AnnotateRowsEventData) eventData).getRowsQuery();
// todo: Cache ANNOTATE_ROWS query with events
// During incremental snapshots, the updates made to the signal table can lead to a case where
// the query stored in the offsets mismatches with the events being dispatched.
//
// IncrementalSnapshotIT#updates performs a series of updates where the pk/aa columns are changed
// i.e. [1,0 to [1,2000] and the ANNOTATE_ROWS event that contains the query specifies this SQL:
// "UPDATE `schema`.`a` SET aa = aa + 2000 WHERE pk > 0 and pk <= 10".
//
// The problem is that signal events do not seem to record a query string in the offsets for MySQL
// but this is recorded for MariaDB, and this causes there to be a mismatch of query string values
// with behavior expected for MySQL. For now, this tests the test to pass until we can better
// understand the root-cause.
if (!Strings.isNullOrBlank(connectorConfig.getSignalingDataCollectionId())) {
final TableId signalDataCollectionId = TableId.parse(connectorConfig.getSignalingDataCollectionId());
if (query.toLowerCase().contains(signalDataCollectionId.toQuotedString('`').toLowerCase())) {
return null;
}
}
return query;
}
@Override

9
debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/IncrementalSnapshotIT.java
View File

@ -28,8 +28,6 @@
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.debezium.config.CommonConnectorConfig;
import io.debezium.config.CommonConnectorConfig.SchemaNameAdjustmentMode;
@ -226,8 +224,6 @@ protected String createTableStatement(String newTable, String copyTable) {
public void updates() throws Exception {
// Testing.Print.enable();
Logger logger = LoggerFactory.getLogger(IncrementalSnapshotIT.class);
populateTable();
startConnector();
@ -250,7 +246,6 @@ public void updates() throws Exception {
x -> ((Struct) x.getValue().value()).getStruct("after").getInt32(valueFieldName()) >= 2000, null);
for (int i = 0; i < expectedRecordCount; i++) {
SourceRecord record = dbChanges.get(i + 1);
logger.info("Record {}: {}", i, record);
final int value = ((Struct) record.value()).getStruct("after").getInt32(valueFieldName());
assertEquals(i + 2000, value);
Object query = ((Struct) record.value()).getStruct("source").get("query");
@ -258,10 +253,6 @@ public void updates() throws Exception {
if (snapshot.equals("false")) {
assertNotNull(query);
}
else if (MySqlTestConnection.isMariaDb()) {
assertNotNull(query);
assertEquals("incremental", snapshot);
}
else {
assertNull(query);
assertEquals("incremental", snapshot);