Package io.debezium.pipeline.source.snapshot.incremental

Class AbstractIncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

java.lang.Object

io.debezium.pipeline.source.snapshot.incremental.AbstractIncrementalSnapshotChangeEventSource<P,T>

All Implemented Interfaces:

IncrementalSnapshotChangeEventSource<P,T>

Direct Known Subclasses:

SignalBasedIncrementalSnapshotChangeEventSource

@NotThreadSafe
public abstract class AbstractIncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>
extends Object
implements IncrementalSnapshotChangeEventSource<P,T>

An incremental snapshot change event source that emits events from a DB log interleaved with snapshot events.

Field Summary

Fields

Modifier and Type	Field	Description
protected ChunkQueryBuilder<T>	chunkQueryBuilder
private final io.debezium.util.Clock	clock
protected final RelationalDatabaseConnectorConfig	connectorConfig
protected IncrementalSnapshotContext<T>	context
private Table	currentTable
private final RelationalDatabaseSchema	databaseSchema
private final DataChangeEventListener<P>	dataListener
protected EventDispatcher<P,T>	dispatcher
protected JdbcConnection	jdbcConnection
private static final org.slf4j.Logger	LOGGER
protected final NotificationService<P,? extends OffsetContext>	notificationService
private final SnapshotProgressListener<P>	progressListener
private long	totalRowsScanned
protected final Map<org.apache.kafka.connect.data.Struct,Object[]>	window

Constructor Summary

Constructors

Constructor	Description
AbstractIncrementalSnapshotChangeEventSource(RelationalDatabaseConnectorConfig config, JdbcConnection jdbcConnection, EventDispatcher<P,T> dispatcher, DatabaseSchema<?> databaseSchema, io.debezium.util.Clock clock, SnapshotProgressListener<P> progressListener, DataChangeEventListener<P> dataChangeEventListener, NotificationService<P,? extends OffsetContext> notificationService)

Method Summary

Modifier and Type	Method	Description
void	addDataCollectionNamesToSnapshot(SignalPayload<P> signalPayload, SnapshotConfiguration snapshotConfiguration)
private void	checkAndProcessStopFlag(P partition, OffsetContext offsetContext)
private boolean	checkSchemaChanges(ResultSet rs)
void	closeWindow(P partition, String id, OffsetContext offsetContext)
protected void	createAndDispatchSchemaChangeEvent(Table newTable, P partition, OffsetContext offsetContext, TableId tableId)	Creates and dispatches a schema change event for the given table.
private boolean	createDataEventsForTable(P partition)	Dispatches the data change events for the records of a single table.
protected void	deduplicateWindow(io.debezium.spi.schema.DataCollectionId dataCollectionId, Object key)
protected abstract void	emitWindowClose(P partition, OffsetContext offsetContext)	Update high watermark for the incremental snapshot chunk
protected abstract void	emitWindowOpen(P partition, OffsetContext offsetContext)	Update low watermark for the incremental snapshot chunk
private List<String>	expandAndDedupeDataCollectionIds(List<String> dataCollectionIds)	Expands the string-based list of data collection ids if supplied using regex to a list of all matching explicit data collection ids.
protected ChangeRecordEmitter<P>	getChangeRecordEmitter(P partition, T dataCollectionId, OffsetContext offsetContext, Object[] row)	Returns a ChangeRecordEmitter producing the change records for the given table row.
protected Optional<String>	getDatabaseName(Object dataCollectionId)	Extract the database name from a data collection ID.
protected String	getSignalTableName(String dataCollectionId)
private Table	getTable(ResultSet rs)
protected String	getTableDDL(TableId tableId)
private io.debezium.util.Threads.Timer	getTableScanLogTimer()
private void	incrementTableRowsScanned(P partition, long rows)
void	init(P partition, OffsetContext offsetContext)
private boolean	isTableInvalid(P partition, OffsetContext offsetContext)
private Object[]	keyFromRow(Object[] row)
private void	nextDataCollection(P partition, OffsetContext offsetContext)
void	pauseSnapshot(P partition, OffsetContext offsetContext)
protected void	postIncrementalSnapshotCompleted()
protected void	postReadChunk(IncrementalSnapshotContext<T> context)
protected void	preIncrementalSnapshotStart()
protected void	preReadChunk(IncrementalSnapshotContext<T> context)
void	processSchemaChange(P partition, OffsetContext offsetContext, io.debezium.spi.schema.DataCollectionId dataCollectionId)
protected void	readChunk(P partition, OffsetContext offsetContext)
private Table	readSchema()
protected Table	readSchemaForTable(TableId tableId)	Reads the schema for the specified table ID.
protected Table	refreshTableSchema(Table table)
void	requestStopSnapshot(P partition, OffsetContext offsetContext, Map<String,Object> additionalData, List<String> dataCollectionIds)
void	rereadChunk(P partition, OffsetContext offsetContext)
void	resumeSnapshot(P partition, OffsetContext offsetContext)
private void	retrieveAndRefreshSchema(TableId tableId, P partition, OffsetContext offsetContext)
private boolean	schemaHistoryIsUpToDate()	Verifies that in-memory representation of the table’s schema is up to date with the table's schema in the database.
protected void	sendEvent(P partition, EventDispatcher<P,T> dispatcher, OffsetContext offsetContext, Object[] row)
protected void	sendWindowEvents(P partition, OffsetContext offsetContext)
protected void	setContext(IncrementalSnapshotContext<T> context)
private void	tableScanCompleted(P partition)
protected void	validateSignalTableConfiguration(List<DataCollection<T>> dataCollections)	Validates that signal table configuration is appropriate for the tables being snapshotted.
private void	verifySchemaUnchanged()	Verifies that table's schema in the database has not changed since it was captured in the previous window
private void	warnAndSkip(P partition, OffsetContext offsetContext, IncrementalSnapshotNotificationService.TableScanCompletionStatus status, String formattedReason, Throwable t)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface io.debezium.pipeline.source.snapshot.incremental.IncrementalSnapshotChangeEventSource

processFilteredEvent, processHeartbeat, processMessage, processTransactionCommittedEvent, processTransactionStartedEvent

Field Details

LOGGER

private static final org.slf4j.Logger LOGGER

connectorConfig

protected final RelationalDatabaseConnectorConfig connectorConfig

clock

private final io.debezium.util.Clock clock

databaseSchema

private final RelationalDatabaseSchema databaseSchema

progressListener

private final SnapshotProgressListener<P extends Partition> progressListener

dataListener

private final DataChangeEventListener<P extends Partition> dataListener

totalRowsScanned

private long totalRowsScanned

currentTable

private Table currentTable

dispatcher

protected EventDispatcher<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId> dispatcher

context

protected IncrementalSnapshotContext<T extends io.debezium.spi.schema.DataCollectionId> context

jdbcConnection

protected JdbcConnection jdbcConnection

chunkQueryBuilder

protected ChunkQueryBuilder<T extends io.debezium.spi.schema.DataCollectionId> chunkQueryBuilder

window

protected final Map<org.apache.kafka.connect.data.Struct,Object[]> window

notificationService

protected final NotificationService<P extends Partition,? extends OffsetContext> notificationService

Constructor Details

AbstractIncrementalSnapshotChangeEventSource

public AbstractIncrementalSnapshotChangeEventSource(RelationalDatabaseConnectorConfig config,
 JdbcConnection jdbcConnection,
 EventDispatcher<P,T> dispatcher,
 DatabaseSchema<?> databaseSchema,
 io.debezium.util.Clock clock,
 SnapshotProgressListener<P> progressListener,
 DataChangeEventListener<P> dataChangeEventListener,
 NotificationService<P,? extends OffsetContext> notificationService)

Method Details

closeWindow

public void closeWindow(P partition,
 String id,
 OffsetContext offsetContext)
                 throws InterruptedException

Specified by:

closeWindow in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

Throws:

InterruptedException

pauseSnapshot

public void pauseSnapshot(P partition,
 OffsetContext offsetContext)

Specified by:

pauseSnapshot in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

resumeSnapshot

public void resumeSnapshot(P partition,
 OffsetContext offsetContext)
                    throws InterruptedException

Specified by:

resumeSnapshot in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

Throws:

InterruptedException

processSchemaChange

public void processSchemaChange(P partition,
 OffsetContext offsetContext,
 io.debezium.spi.schema.DataCollectionId dataCollectionId)
                         throws InterruptedException

Specified by:

processSchemaChange in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

Throws:

InterruptedException

rereadChunk

public void rereadChunk(P partition,
 OffsetContext offsetContext)
                 throws InterruptedException

Throws:

InterruptedException

getSignalTableName

protected String getSignalTableName(String dataCollectionId)

sendWindowEvents

protected void sendWindowEvents(P partition,
 OffsetContext offsetContext)
                         throws InterruptedException

Throws:

InterruptedException

sendEvent

protected void sendEvent(P partition,
 EventDispatcher<P,T> dispatcher,
 OffsetContext offsetContext,
 Object[] row)
                  throws InterruptedException

Throws:

InterruptedException

getChangeRecordEmitter

protected ChangeRecordEmitter<P> getChangeRecordEmitter(P partition,
 T dataCollectionId,
 OffsetContext offsetContext,
 Object[] row)

Returns a ChangeRecordEmitter producing the change records for the given table row.

deduplicateWindow

protected void deduplicateWindow(io.debezium.spi.schema.DataCollectionId dataCollectionId,
 Object key)

emitWindowOpen

protected abstract void emitWindowOpen(P partition,
 OffsetContext offsetContext)
                                throws SQLException

Update low watermark for the incremental snapshot chunk

Throws:

SQLException

emitWindowClose

protected abstract void emitWindowClose(P partition,
 OffsetContext offsetContext)
                                 throws Exception

Update high watermark for the incremental snapshot chunk

Throws:

Exception

init

public void init(P partition,
 OffsetContext offsetContext)

Specified by:

init in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

readChunk

protected void readChunk(P partition,
 OffsetContext offsetContext)
                  throws InterruptedException

Throws:

InterruptedException

isTableInvalid

private boolean isTableInvalid(P partition,
 OffsetContext offsetContext)

warnAndSkip

private void warnAndSkip(P partition,
 OffsetContext offsetContext,
 IncrementalSnapshotNotificationService.TableScanCompletionStatus status,
 String formattedReason,
 Throwable t)

retrieveAndRefreshSchema

private void retrieveAndRefreshSchema(TableId tableId,
 P partition,
 OffsetContext offsetContext)
                               throws SQLException,
InterruptedException

Throws:

SQLException

InterruptedException

schemaHistoryIsUpToDate

private boolean schemaHistoryIsUpToDate()

Verifies that in-memory representation of the table’s schema is up to date with the table's schema in the database.

Verification is a two step process:

1. Save table's schema from the database to the context
2. Verify schema hasn't changed in the following window. If schema has changed repeat the process

Two step process allows to wait for the connector to receive the DDL event in the binlog stream and update the in-memory representation of the table’s schema.

Verification is done at the beginning of the incremental snapshot and on every schema change during the snapshotting.

verifySchemaUnchanged

private void verifySchemaUnchanged()

Verifies that table's schema in the database has not changed since it was captured in the previous window

readSchema

private Table readSchema()

nextDataCollection

private void nextDataCollection(P partition,
 OffsetContext offsetContext)

addDataCollectionNamesToSnapshot

public void addDataCollectionNamesToSnapshot(SignalPayload<P> signalPayload,
 SnapshotConfiguration snapshotConfiguration)
                                      throws InterruptedException

Specified by:

addDataCollectionNamesToSnapshot in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

Throws:

InterruptedException

validateSignalTableConfiguration

protected void validateSignalTableConfiguration(List<DataCollection<T>> dataCollections)

Validates that signal table configuration is appropriate for the tables being snapshotted. For multi-database connectors with a single signal table, warns if tables from different databases are being snapshotted.

getDatabaseName

protected Optional<String> getDatabaseName(Object dataCollectionId)

Extract the database name from a data collection ID. Default implementation handles TableId, subclasses may override.

requestStopSnapshot

public void requestStopSnapshot(P partition,
 OffsetContext offsetContext,
 Map<String,Object> additionalData,
 List<String> dataCollectionIds)

Specified by:

requestStopSnapshot in interface IncrementalSnapshotChangeEventSource<P extends Partition,T extends io.debezium.spi.schema.DataCollectionId>

checkAndProcessStopFlag

private void checkAndProcessStopFlag(P partition,
 OffsetContext offsetContext)

expandAndDedupeDataCollectionIds

private List<String> expandAndDedupeDataCollectionIds(List<String> dataCollectionIds)

Expands the string-based list of data collection ids if supplied using regex to a list of all matching explicit data collection ids.

createDataEventsForTable

private boolean createDataEventsForTable(P partition)
                                  throws SQLException

Dispatches the data change events for the records of a single table.

Throws:

SQLException

checkSchemaChanges

private boolean checkSchemaChanges(ResultSet rs)
                            throws SQLException

Throws:

SQLException

getTable

private Table getTable(ResultSet rs)
                throws SQLException

Throws:

SQLException

incrementTableRowsScanned

private void incrementTableRowsScanned(P partition,
 long rows)

tableScanCompleted

private void tableScanCompleted(P partition)

getTableScanLogTimer

private io.debezium.util.Threads.Timer getTableScanLogTimer()

keyFromRow

private Object[] keyFromRow(Object[] row)

setContext

protected void setContext(IncrementalSnapshotContext<T> context)

preIncrementalSnapshotStart

protected void preIncrementalSnapshotStart()

preReadChunk

protected void preReadChunk(IncrementalSnapshotContext<T> context)

postReadChunk

protected void postReadChunk(IncrementalSnapshotContext<T> context)

postIncrementalSnapshotCompleted

protected void postIncrementalSnapshotCompleted()

refreshTableSchema

protected Table refreshTableSchema(Table table)
                            throws SQLException

Throws:

SQLException

readSchemaForTable

protected Table readSchemaForTable(TableId tableId)
                            throws SQLException

Reads the schema for the specified table ID. This method can be overridden if custom schema retrieval logic is required.

Throws:

SQLException

createAndDispatchSchemaChangeEvent

protected void createAndDispatchSchemaChangeEvent(Table newTable,
 P partition,
 OffsetContext offsetContext,
 TableId tableId)
                                           throws SQLException

Creates and dispatches a schema change event for the given table. This method can be overridden if custom event creation or dispatching is required.

Throws:

SQLException

getTableDDL

protected String getTableDDL(TableId tableId)
                      throws SQLException

Throws:

SQLException