DBZ-777 Misc. adjustments
* Avoiding repeated parsing of table id * Making 'inconsistentSchemaHandler' field final in EventDispatcher * Not exposing synchronizeTableSchema() on SPI level
This commit is contained in:
parent
780545d0d2
commit
63250d0c13
@ -43,6 +43,7 @@ public class PostgresChangeRecordEmitter extends RelationalChangeRecordEmitter {
|
|||||||
private final PostgresSchema schema;
|
private final PostgresSchema schema;
|
||||||
private final PostgresConnectorConfig connectorConfig;
|
private final PostgresConnectorConfig connectorConfig;
|
||||||
private final PostgresConnection connection;
|
private final PostgresConnection connection;
|
||||||
|
private final TableId tableId;
|
||||||
|
|
||||||
public PostgresChangeRecordEmitter(OffsetContext offset, Clock clock, PostgresConnectorConfig connectorConfig, PostgresSchema schema, PostgresConnection connection, ReplicationMessage message) {
|
public PostgresChangeRecordEmitter(OffsetContext offset, Clock clock, PostgresConnectorConfig connectorConfig, PostgresSchema schema, PostgresConnection connection, ReplicationMessage message) {
|
||||||
super(offset, clock);
|
super(offset, clock);
|
||||||
@ -51,6 +52,9 @@ public PostgresChangeRecordEmitter(OffsetContext offset, Clock clock, PostgresCo
|
|||||||
this.message = message;
|
this.message = message;
|
||||||
this.connectorConfig = connectorConfig;
|
this.connectorConfig = connectorConfig;
|
||||||
this.connection = connection;
|
this.connection = connection;
|
||||||
|
|
||||||
|
this.tableId = PostgresSchema.parse(message.getTable());
|
||||||
|
Objects.requireNonNull(tableId);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -68,10 +72,13 @@ protected Operation getOperation() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Object[] getOldColumnValues() {
|
public void emitChangeRecords(DataCollectionSchema schema, Receiver receiver) throws InterruptedException {
|
||||||
final TableId tableId = PostgresSchema.parse(message.getTable());
|
schema = synchronizeTableSchema(schema);
|
||||||
Objects.requireNonNull(tableId);
|
super.emitChangeRecords(schema, receiver);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected Object[] getOldColumnValues() {
|
||||||
try {
|
try {
|
||||||
switch (getOperation()) {
|
switch (getOperation()) {
|
||||||
case CREATE:
|
case CREATE:
|
||||||
@ -89,9 +96,6 @@ protected Object[] getOldColumnValues() {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Object[] getNewColumnValues() {
|
protected Object[] getNewColumnValues() {
|
||||||
final TableId tableId = PostgresSchema.parse(message.getTable());
|
|
||||||
Objects.requireNonNull(tableId);
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
switch (getOperation()) {
|
switch (getOperation()) {
|
||||||
case CREATE:
|
case CREATE:
|
||||||
@ -107,8 +111,7 @@ protected Object[] getNewColumnValues() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
private DataCollectionSchema synchronizeTableSchema(DataCollectionSchema tableSchema) {
|
||||||
public DataCollectionSchema synchronizeTableSchema(DataCollectionSchema tableSchema) {
|
|
||||||
final boolean metadataInMessage = message.hasTypeMetadata();
|
final boolean metadataInMessage = message.hasTypeMetadata();
|
||||||
final TableId tableId = (TableId) tableSchema.id();
|
final TableId tableId = (TableId) tableSchema.id();
|
||||||
final Table table = schema.tableFor(tableId);
|
final Table table = schema.tableFor(tableId);
|
||||||
|
@ -145,8 +145,9 @@ public void start(Configuration config) {
|
|||||||
schema,
|
schema,
|
||||||
queue,
|
queue,
|
||||||
connectorConfig.getTableFilters().dataCollectionFilter(),
|
connectorConfig.getTableFilters().dataCollectionFilter(),
|
||||||
DataChangeEvent::new);
|
DataChangeEvent::new,
|
||||||
dispatcher.setInconsistentSchemaHandler(PostgresChangeRecordEmitter::updateSchema);
|
PostgresChangeRecordEmitter::updateSchema
|
||||||
|
);
|
||||||
|
|
||||||
coordinator = new ChangeEventSourceCoordinator(
|
coordinator = new ChangeEventSourceCoordinator(
|
||||||
previousOffset,
|
previousOffset,
|
||||||
|
@ -6,7 +6,6 @@
|
|||||||
package io.debezium.connector.postgresql.snapshot;
|
package io.debezium.connector.postgresql.snapshot;
|
||||||
|
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.util.Map;
|
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
|
|
||||||
@ -22,11 +21,8 @@
|
|||||||
*/
|
*/
|
||||||
public class ExportedSnapshotter implements Snapshotter {
|
public class ExportedSnapshotter implements Snapshotter {
|
||||||
|
|
||||||
private Map<TableId, String> snapshotOverrides;
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void init(PostgresConnectorConfig config, OffsetState sourceInfo, SlotState slotState) {
|
public void init(PostgresConnectorConfig config, OffsetState sourceInfo, SlotState slotState) {
|
||||||
this.snapshotOverrides = config.getSnapshotSelectOverridesByTable();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -46,13 +42,8 @@ public boolean exportSnapshot() {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Optional<String> buildSnapshotQuery(TableId tableId) {
|
public Optional<String> buildSnapshotQuery(TableId tableId) {
|
||||||
if (snapshotOverrides.containsKey(tableId)) {
|
|
||||||
return Optional.of(snapshotOverrides.get(tableId));
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
return Optional.of("select * from " + tableId.toDoubleQuotedString());
|
return Optional.of("select * from " + tableId.toDoubleQuotedString());
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Optional<String> snapshotTableLockingStatement(Duration lockTimeout, Set<TableId> tableIds) {
|
public Optional<String> snapshotTableLockingStatement(Duration lockTimeout, Set<TableId> tableIds) {
|
||||||
|
@ -5,7 +5,6 @@
|
|||||||
*/
|
*/
|
||||||
package io.debezium.connector.postgresql.snapshot;
|
package io.debezium.connector.postgresql.snapshot;
|
||||||
|
|
||||||
import java.util.Map;
|
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
|
|
||||||
import io.debezium.connector.postgresql.PostgresConnectorConfig;
|
import io.debezium.connector.postgresql.PostgresConnectorConfig;
|
||||||
@ -16,26 +15,16 @@
|
|||||||
|
|
||||||
public abstract class QueryingSnapshotter implements Snapshotter {
|
public abstract class QueryingSnapshotter implements Snapshotter {
|
||||||
|
|
||||||
private Map<TableId, String> snapshotOverrides;
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void init(PostgresConnectorConfig config, OffsetState sourceInfo, SlotState slotState) {
|
public void init(PostgresConnectorConfig config, OffsetState sourceInfo, SlotState slotState) {
|
||||||
this.snapshotOverrides = config.getSnapshotSelectOverridesByTable();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Optional<String> buildSnapshotQuery(TableId tableId) {
|
public Optional<String> buildSnapshotQuery(TableId tableId) {
|
||||||
if (snapshotOverrides.containsKey(tableId)) {
|
|
||||||
return Optional.of(snapshotOverrides.get(tableId));
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
// DBZ-298 Quoting name in case it has been quoted originally; it doesn't do harm if it hasn't been quoted
|
// DBZ-298 Quoting name in case it has been quoted originally; it doesn't do harm if it hasn't been quoted
|
||||||
StringBuilder q = new StringBuilder();
|
StringBuilder q = new StringBuilder();
|
||||||
q.append("SELECT * FROM ");
|
q.append("SELECT * FROM ");
|
||||||
q.append(tableId.toDoubleQuotedString());
|
q.append(tableId.toDoubleQuotedString());
|
||||||
return Optional.of(q.toString());
|
return Optional.of(q.toString());
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
@ -57,7 +57,7 @@ public class EventDispatcher<T extends DataCollectionId> {
|
|||||||
private final Heartbeat heartbeat;
|
private final Heartbeat heartbeat;
|
||||||
private DataChangeEventListener eventListener = DataChangeEventListener.NO_OP;
|
private DataChangeEventListener eventListener = DataChangeEventListener.NO_OP;
|
||||||
private final boolean emitTombstonesOnDelete;
|
private final boolean emitTombstonesOnDelete;
|
||||||
private InconsistentSchemaHandler<T> inconsistentSchemaHandler = this::errorOnMissingSchema;
|
private final InconsistentSchemaHandler<T> inconsistentSchemaHandler;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Change event receiver for events dispatched from a streaming change event source.
|
* Change event receiver for events dispatched from a streaming change event source.
|
||||||
@ -67,6 +67,11 @@ public class EventDispatcher<T extends DataCollectionId> {
|
|||||||
public EventDispatcher(CommonConnectorConfig connectorConfig, TopicSelector<T> topicSelector,
|
public EventDispatcher(CommonConnectorConfig connectorConfig, TopicSelector<T> topicSelector,
|
||||||
DatabaseSchema<T> schema, ChangeEventQueue<DataChangeEvent> queue, DataCollectionFilter<T> filter,
|
DatabaseSchema<T> schema, ChangeEventQueue<DataChangeEvent> queue, DataCollectionFilter<T> filter,
|
||||||
ChangeEventCreator changeEventCreator) {
|
ChangeEventCreator changeEventCreator) {
|
||||||
|
this(connectorConfig, topicSelector, schema, queue, filter, changeEventCreator, null);
|
||||||
|
}
|
||||||
|
public EventDispatcher(CommonConnectorConfig connectorConfig, TopicSelector<T> topicSelector,
|
||||||
|
DatabaseSchema<T> schema, ChangeEventQueue<DataChangeEvent> queue, DataCollectionFilter<T> filter,
|
||||||
|
ChangeEventCreator changeEventCreator, InconsistentSchemaHandler<T> inconsistentSchemaHandler) {
|
||||||
this.topicSelector = topicSelector;
|
this.topicSelector = topicSelector;
|
||||||
this.schema = schema;
|
this.schema = schema;
|
||||||
this.historizedSchema = schema instanceof HistorizedDatabaseSchema
|
this.historizedSchema = schema instanceof HistorizedDatabaseSchema
|
||||||
@ -77,6 +82,7 @@ public EventDispatcher(CommonConnectorConfig connectorConfig, TopicSelector<T> t
|
|||||||
this.changeEventCreator = changeEventCreator;
|
this.changeEventCreator = changeEventCreator;
|
||||||
this.streamingReceiver = new StreamingChangeRecordReceiver();
|
this.streamingReceiver = new StreamingChangeRecordReceiver();
|
||||||
this.emitTombstonesOnDelete = connectorConfig.isEmitTombstoneOnDelete();
|
this.emitTombstonesOnDelete = connectorConfig.isEmitTombstoneOnDelete();
|
||||||
|
this.inconsistentSchemaHandler = inconsistentSchemaHandler != null ? inconsistentSchemaHandler : this::errorOnMissingSchema;
|
||||||
|
|
||||||
heartbeat = Heartbeat.create(connectorConfig.getConfig(), topicSelector.getHeartbeatTopic(),
|
heartbeat = Heartbeat.create(connectorConfig.getConfig(), topicSelector.getHeartbeatTopic(),
|
||||||
connectorConfig.getLogicalName());
|
connectorConfig.getLogicalName());
|
||||||
@ -131,7 +137,6 @@ public void dispatchDataChangeEvent(T dataCollectionId, ChangeRecordEmitter chan
|
|||||||
}
|
}
|
||||||
dataCollectionSchema = replacementSchema.get();
|
dataCollectionSchema = replacementSchema.get();
|
||||||
}
|
}
|
||||||
dataCollectionSchema = changeRecordEmitter.synchronizeTableSchema(dataCollectionSchema);
|
|
||||||
|
|
||||||
changeRecordEmitter.emitChangeRecords(dataCollectionSchema, new Receiver() {
|
changeRecordEmitter.emitChangeRecords(dataCollectionSchema, new Receiver() {
|
||||||
|
|
||||||
@ -151,10 +156,6 @@ public void changeRecord(DataCollectionSchema schema, Operation operation, Objec
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setInconsistentSchemaHandler(InconsistentSchemaHandler<T> inconsistentSchemaHandler) {
|
|
||||||
this.inconsistentSchemaHandler = inconsistentSchemaHandler;
|
|
||||||
}
|
|
||||||
|
|
||||||
public Optional<DataCollectionSchema> errorOnMissingSchema(T dataCollectionId, ChangeRecordEmitter changeRecordEmitter) {
|
public Optional<DataCollectionSchema> errorOnMissingSchema(T dataCollectionId, ChangeRecordEmitter changeRecordEmitter) {
|
||||||
eventListener.onErroneousEvent("source = " + dataCollectionId);
|
eventListener.onErroneousEvent("source = " + dataCollectionId);
|
||||||
throw new IllegalArgumentException("No metadata registered for captured table " + dataCollectionId);
|
throw new IllegalArgumentException("No metadata registered for captured table " + dataCollectionId);
|
||||||
|
@ -29,10 +29,6 @@ public interface ChangeRecordEmitter {
|
|||||||
*/
|
*/
|
||||||
OffsetContext getOffset();
|
OffsetContext getOffset();
|
||||||
|
|
||||||
default DataCollectionSchema synchronizeTableSchema(DataCollectionSchema dataCollectionSchema) {
|
|
||||||
return dataCollectionSchema;
|
|
||||||
}
|
|
||||||
|
|
||||||
public interface Receiver {
|
public interface Receiver {
|
||||||
void changeRecord(DataCollectionSchema schema, Operation operation, Object key, Struct value, OffsetContext offset) throws InterruptedException;
|
void changeRecord(DataCollectionSchema schema, Operation operation, Object key, Struct value, OffsetContext offset) throws InterruptedException;
|
||||||
}
|
}
|
||||||
|
@ -51,8 +51,6 @@
|
|||||||
*
|
*
|
||||||
* @author Gunnar Morling
|
* @author Gunnar Morling
|
||||||
*/
|
*/
|
||||||
// TODO Mostly, this should be usable for Postgres as well; only the aspect of managing the schema history will have to
|
|
||||||
// be made optional based on the connector
|
|
||||||
public abstract class RelationalSnapshotChangeEventSource implements SnapshotChangeEventSource {
|
public abstract class RelationalSnapshotChangeEventSource implements SnapshotChangeEventSource {
|
||||||
|
|
||||||
private static final Logger LOGGER = LoggerFactory.getLogger(RelationalSnapshotChangeEventSource.class);
|
private static final Logger LOGGER = LoggerFactory.getLogger(RelationalSnapshotChangeEventSource.class);
|
||||||
|
Loading…
Reference in New Issue
Block a user