Skip to content

Commit 1f066df

Browse files
mikekamornikovmikekamornikov
committed
CXP-2779: extracted HistoryRecordProcessor logic, added SnapshotAware SchemaHistory implementation, added SchemaHistorySnapshot with NOOP (default) and inmemory implementation
1 parent 7c8892c commit 1f066df

27 files changed

Lines changed: 846 additions & 111 deletions

File tree

‎debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlConnectorConfig.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import io.debezium.relational.TableId;
3535
import io.debezium.relational.Tables.TableFilter;
3636
import io.debezium.relational.history.HistoryRecordComparator;
37+
import io.debezium.relational.history.snapshot.SchemaPartitioner;
3738
import io.debezium.schema.DefaultTopicNamingStrategy;
3839
import io.debezium.util.Collect;
3940

@@ -1164,6 +1165,11 @@ protected HistoryRecordComparator getHistoryRecordComparator() {
11641165
return new MySqlHistoryRecordComparator(gtidSourceFilter());
11651166
}
11661167

1168+
@Override
1169+
protected SchemaPartitioner getSchemaPartitioner() {
1170+
return SchemaPartitioner.SINGLE_PARTITION;
1171+
}
1172+
11671173
public static boolean isBuiltInDatabase(String databaseName) {
11681174
if (databaseName == null) {
11691175
return false;

‎debezium-connector-mysql/src/test/java/io/debezium/relational/history/KafkaSchemaHistoryTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ public void shouldStartWithEmptyTopicAndStoreDataAndRecoverAllState() throws Exc
120120
}
121121

122122
private void testHistoryTopicContent(String topicName, boolean skipUnparseableDDL) {
123-
interceptor = new LogInterceptor(KafkaSchemaHistory.class);
123+
interceptor = new LogInterceptor(HistoryRecordProcessor.class);
124124
// Start up the history ...
125125
Configuration config = Configuration.create()
126126
.with(KafkaSchemaHistory.BOOTSTRAP_SERVERS, kafka.brokerList())

‎debezium-connector-mysql/src/test/java/io/debezium/relational/history/snapshot/AbstractSchemaHistorySnapshotTest.java‎

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
package io.debezium.relational.history.snapshot;
2+
3+
import static org.junit.Assert.assertEquals;
4+
import static org.junit.Assert.assertNull;
5+
import static org.junit.Assert.fail;
6+
7+
import java.util.Map;
8+
9+
import org.junit.After;
10+
import org.junit.Before;
11+
import org.junit.Test;
12+
13+
import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
14+
import io.debezium.relational.Tables;
15+
import io.debezium.relational.ddl.DdlParser;
16+
import io.debezium.relational.history.MemorySchemaHistory;
17+
import io.debezium.relational.history.SchemaHistory;
18+
19+
public abstract class AbstractSchemaHistorySnapshotTest {
20+
21+
protected SchemaHistorySnapshot snapshot;
22+
protected SchemaHistory history;
23+
protected Map<String, Object> source1;
24+
protected Map<String, Object> source2;
25+
protected Tables s1_t0, s1_t1, s1_t2, s1_t3;
26+
protected Tables s2_t0, s2_t1, s2_t2, s2_t3;
27+
protected DdlParser parser;
28+
29+
@Before
30+
public void beforeEach() {
31+
history = new MemorySchemaHistory();
32+
snapshot = createHistorySnapshot();
33+
34+
source1 = server("abc");
35+
source2 = server("xyz");
36+
37+
parser = new MySqlAntlrDdlParser();
38+
s1_t0 = new Tables();
39+
s1_t1 = new Tables();
40+
s1_t2 = new Tables();
41+
s1_t3 = new Tables();
42+
43+
record(1, "CREATE TABLE foo ( first VARCHAR(22) NOT NULL );", s1_t3, s1_t2, s1_t1, s1_t0);
44+
record(23, "CREATE TABLE\nperson ( name VARCHAR(22) NOT NULL );", s1_t3, s1_t2, s1_t1);
45+
record(30, "CREATE TABLE address\n( street VARCHAR(22) NOT NULL );", s1_t3, s1_t2);
46+
record(32, "ALTER TABLE address ADD city VARCHAR(22) NOT NULL;", s1_t3);
47+
48+
snapshot.save(source1, position(32), s1_t3);
49+
snapshot.save(source1, position(30), s1_t2);
50+
snapshot.save(source1, position(23), s1_t1);
51+
snapshot.save(source1, position(1), s1_t0);
52+
53+
s2_t0 = new Tables();
54+
s2_t1 = new Tables();
55+
s2_t2 = new Tables();
56+
s2_t3 = new Tables();
57+
58+
record(2, "CREATE TABLE foo1 ( first VARCHAR(22) NOT NULL );", s2_t3, s2_t2, s2_t1, s2_t0);
59+
record(12, "CREATE TABLE\nperson1 ( name VARCHAR(22) NOT NULL );", s2_t3, s2_t2, s2_t1);
60+
record(16, "CREATE TABLE address1\n( street VARCHAR(22) NOT NULL );", s2_t3, s2_t2);
61+
record(99, "ALTER TABLE address1 ADD city VARCHAR(22) NOT NULL;", s2_t3);
62+
63+
snapshot.save(source2, position(55), s2_t3);
64+
snapshot.save(source2, position(16), s2_t2);
65+
snapshot.save(source2, position(12), s2_t1);
66+
snapshot.save(source2, position(2), s2_t0);
67+
}
68+
69+
@After
70+
public void afterEach() {
71+
}
72+
73+
protected abstract SchemaHistorySnapshot createHistorySnapshot();
74+
75+
protected Map<String, Object> server(String serverName) {
76+
return Map.of("server", serverName);
77+
}
78+
79+
protected Map<String, Object> position(long pos) {
80+
return Map.of("file", "mysql-bin-changelog.000011", "pos", pos);
81+
}
82+
83+
protected void record(long pos, String ddl, Tables... update) {
84+
try {
85+
history.record(source1, position(pos), "db", ddl);
86+
}
87+
catch (Throwable t) {
88+
fail(t.getMessage());
89+
}
90+
for (Tables tables : update) {
91+
if (tables != null) {
92+
parser.setCurrentSchema("db");
93+
parser.parse(ddl, tables);
94+
}
95+
}
96+
}
97+
98+
@Test
99+
public void shouldRecordSnapshotsAndRecoverToClosest() {
100+
assertNull(snapshot.read(source1, null));
101+
assertNull(snapshot.read(null, position(31)));
102+
assertNull(snapshot.read(source1, position(31)));
103+
assertEquals(s1_t2, snapshot.read(source1, position(30)));
104+
105+
assertNull(snapshot.read(source2, position(30)));
106+
assertEquals(s2_t1, snapshot.read(source2, position(12)));
107+
108+
assertNull(snapshot.findClosest(null, position(31)));
109+
assertNull(snapshot.findClosest(source1, null));
110+
assertEquals(position(30), snapshot.findClosest(source1, position(31)));
111+
assertEquals(position(32), snapshot.findClosest(source1, position(99)));
112+
113+
assertNull(snapshot.findClosest(source2, position(1)));
114+
assertEquals(position(55), snapshot.findClosest(source2, position(99)));
115+
}
116+
}

‎debezium-connector-mysql/src/test/java/io/debezium/relational/history/snapshot/MemorySchemaHistorySnapshotTest.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package io.debezium.relational.history.snapshot;
2+
3+
public class MemorySchemaHistorySnapshotTest extends AbstractSchemaHistorySnapshotTest {
4+
protected SchemaHistorySnapshot createHistorySnapshot() {
5+
return new MemorySchemaHistorySnapshot();
6+
}
7+
}

‎debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/OracleConnectorConfig.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747
import io.debezium.relational.TableId;
4848
import io.debezium.relational.Tables.TableFilter;
4949
import io.debezium.relational.history.HistoryRecordComparator;
50+
import io.debezium.relational.history.snapshot.SchemaPartitioner;
5051
import io.debezium.util.Strings;
5152

5253
/**
@@ -829,6 +830,11 @@ protected HistoryRecordComparator getHistoryRecordComparator() {
829830
return getAdapter().getHistoryRecordComparator();
830831
}
831832

833+
@Override
834+
protected SchemaPartitioner getSchemaPartitioner() {
835+
return SchemaPartitioner.SINGLE_PARTITION;
836+
}
837+
832838
/**
833839
* Defines modes of representation of {@code interval} datatype
834840
*/

‎debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnectorConfig.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import io.debezium.relational.TableId;
3434
import io.debezium.relational.Tables.TableFilter;
3535
import io.debezium.relational.history.HistoryRecordComparator;
36+
import io.debezium.relational.history.snapshot.SchemaPartitioner;
3637
import io.debezium.spi.schema.DataCollectionId;
3738
import io.debezium.util.Strings;
3839

@@ -493,6 +494,15 @@ protected boolean isPositionAtOrBefore(Document recorded, Document desired) {
493494
};
494495
}
495496

497+
@Override
498+
protected SchemaPartitioner getSchemaPartitioner() {
499+
return (source) -> {
500+
return TableFilter.fromPredicate((tableId) -> {
501+
return tableId.catalog().equals(((SqlServerPartition)source).getDatabaseName());
502+
});
503+
};
504+
}
505+
496506
@Override
497507
public String getContextName() {
498508
return Module.contextName();

‎debezium-connector-sqlserver/src/test/java/io/debezium/connector/sqlserver/SqlServerConnectorIT.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@
6767
import io.debezium.relational.Tables;
6868
import io.debezium.relational.ddl.DdlParser;
6969
import io.debezium.relational.history.HistoryRecordComparator;
70+
import io.debezium.relational.history.HistoryRecordProcessorProvider;
7071
import io.debezium.relational.history.SchemaHistory;
7172
import io.debezium.relational.history.SchemaHistoryException;
7273
import io.debezium.relational.history.SchemaHistoryListener;
@@ -2907,8 +2908,8 @@ public boolean exists() {
29072908

29082909
@Override
29092910
public void configure(Configuration config, HistoryRecordComparator comparator,
2910-
SchemaHistoryListener listener, boolean useCatalogBeforeSchema) {
2911-
delegate.configure(config, comparator, listener, useCatalogBeforeSchema);
2911+
SchemaHistoryListener listener, HistoryRecordProcessorProvider processorProvider) {
2912+
delegate.configure(config, comparator, listener, processorProvider);
29122913
}
29132914

29142915
@Override

‎debezium-core/src/main/java/io/debezium/relational/HistorizedRelationalDatabaseConnectorConfig.java‎

Lines changed: 31 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,14 @@
2121
import io.debezium.relational.Selectors.TableIdToStringMapper;
2222
import io.debezium.relational.Tables.TableFilter;
2323
import io.debezium.relational.history.HistoryRecordComparator;
24+
import io.debezium.relational.history.HistoryRecordProcessor;
25+
import io.debezium.relational.history.HistoryRecordProcessorProvider;
2426
import io.debezium.relational.history.SchemaHistory;
27+
import io.debezium.relational.history.SchemaHistoryListener;
2528
import io.debezium.relational.history.SchemaHistoryMetrics;
29+
import io.debezium.relational.history.SnapshotAwareSchemaHistory;
30+
import io.debezium.relational.history.snapshot.SchemaHistorySnapshot;
31+
import io.debezium.relational.history.snapshot.SchemaPartitioner;
2632

2733
/**
2834
* Configuration options shared across the relational CDC connectors which use a persistent database schema history.
@@ -34,6 +40,7 @@ public abstract class HistorizedRelationalDatabaseConnectorConfig extends Relati
3440
protected static final int DEFAULT_SNAPSHOT_FETCH_SIZE = 2_000;
3541

3642
private static final String DEFAULT_SCHEMA_HISTORY = "io.debezium.storage.kafka.history.KafkaSchemaHistory";
43+
private static final String DEFAULT_SCHEMA_HISTORY_SNAPSHOT = "io.debezium.relational.history.snapshot.SchemaHistorySnapshot.NOOP";
3744

3845
private final boolean useCatalogBeforeSchema;
3946
private final Class<? extends SourceConnector> connectorClass;
@@ -58,6 +65,17 @@ public abstract class HistorizedRelationalDatabaseConnectorConfig extends Relati
5865
+ SchemaHistory.CONFIGURATION_FIELD_PREFIX_STRING + "' string.")
5966
.withDefault(DEFAULT_SCHEMA_HISTORY);
6067

68+
public static final Field SCHEMA_HISTORY_SNAPSHOT = Field.create("schema.history.internal.snapshot")
69+
.withDisplayName("Database schema history snapshot class")
70+
.withType(Type.CLASS)
71+
.withWidth(Width.LONG)
72+
.withImportance(Importance.LOW)
73+
.withInvisibleRecommender()
74+
.withDescription("The name of the SchemaHistorySnapshot class that should be used to store and recover database schema. "
75+
+ "The configuration properties for the history snapshot are prefixed with the '"
76+
+ SchemaHistorySnapshot.CONFIGURATION_FIELD_PREFIX_STRING + "' string.")
77+
.withDefault(DEFAULT_SCHEMA_HISTORY_SNAPSHOT);
78+
6179
public static final Field SKIP_UNPARSEABLE_DDL_STATEMENTS = SchemaHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS;
6280

6381
public static final Field STORE_ONLY_CAPTURED_TABLES_DDL = SchemaHistory.STORE_ONLY_CAPTURED_TABLES_DDL;
@@ -133,9 +151,19 @@ public SchemaHistory getSchemaHistory() {
133151
.withDefault(SchemaHistory.INTERNAL_CONNECTOR_ID, logicalName)
134152
.build();
135153

154+
SchemaHistorySnapshot schemaSnapshot = config.getInstance(SCHEMA_HISTORY_SNAPSHOT, SchemaHistorySnapshot.class);
155+
if (schemaSnapshot == null) {
156+
throw new ConnectException("Unable to instantiate the database schema history snapshot class " +
157+
config.getString(SCHEMA_HISTORY_SNAPSHOT));
158+
}
159+
160+
schemaHistory = new SnapshotAwareSchemaHistory(schemaHistory, schemaSnapshot, getSchemaPartitioner());
161+
136162
HistoryRecordComparator historyComparator = getHistoryRecordComparator();
137-
schemaHistory.configure(schemaHistoryConfig, historyComparator,
138-
new SchemaHistoryMetrics(this, multiPartitionMode()), useCatalogBeforeSchema()); // validates
163+
SchemaHistoryListener historyListener = new SchemaHistoryMetrics(this, multiPartitionMode());
164+
HistoryRecordProcessorProvider processorProvider = (o, s, p) -> new HistoryRecordProcessor(o, s, p, schemaHistoryConfig, historyComparator, historyListener,
165+
useCatalogBeforeSchema());
166+
schemaHistory.configure(schemaHistoryConfig, historyComparator, historyListener, processorProvider); // validates
139167

140168
return schemaHistory;
141169
}
@@ -177,4 +205,5 @@ public boolean storeOnlyCapturedDatabases() {
177205
*/
178206
protected abstract HistoryRecordComparator getHistoryRecordComparator();
179207

208+
protected abstract SchemaPartitioner getSchemaPartitioner();
180209
}

‎debezium-core/src/main/java/io/debezium/relational/ddl/DdlParser.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,4 +40,31 @@ public interface DdlParser {
4040
DdlChanges getDdlChanges();
4141

4242
SystemVariables systemVariables();
43+
44+
DdlParser NOOP = new DdlParser() {
45+
@Override
46+
public void parse(String ddlContent, Tables databaseTables) {
47+
48+
}
49+
50+
@Override
51+
public void setCurrentDatabase(String databaseName) {
52+
53+
}
54+
55+
@Override
56+
public void setCurrentSchema(String schemaName) {
57+
58+
}
59+
60+
@Override
61+
public DdlChanges getDdlChanges() {
62+
return null;
63+
}
64+
65+
@Override
66+
public SystemVariables systemVariables() {
67+
return null;
68+
}
69+
};
4370
}

0 commit comments

Comments
 (0)