Package io.debezium.pipeline

Class AbstractBlockingSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

java.lang.Object

io.debezium.embedded.AbstractConnectorTest

io.debezium.embedded.async.AbstractAsyncEngineConnectorTest

io.debezium.pipeline.source.snapshot.incremental.AbstractSnapshotTest<T>

io.debezium.pipeline.AbstractBlockingSnapshotTest<T>

All Implemented Interfaces:

Testing

public abstract class AbstractBlockingSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>
extends AbstractSnapshotTest<T>

Nested Class Summary

Nested classes/interfaces inherited from class io.debezium.embedded.AbstractConnectorTest

AbstractConnectorTest.SourceRecords

Nested classes/interfaces inherited from interface io.debezium.util.Testing

Testing.Debug, Testing.Files, Testing.InterruptableFunction, Testing.Network, Testing.Print, Testing.Timer

Field Summary

Fields

Modifier and Type	Field	Description
protected static final int	ROW_COUNT
private int	signalingRecords

Fields inherited from class io.debezium.pipeline.source.snapshot.incremental.AbstractSnapshotTest

PARTITION_NO, SCHEMA_HISTORY_PATH, SERVER_NAME, signalsFile

Fields inherited from class io.debezium.embedded.AbstractConnectorTest

consumedLines, engine, isEngineRunning, logger, logTestName, OFFSET_STORE_PATH, pollTimeoutInMs, requiresAssemblyProfileClassRule, requiresAssemblyProfileRule, skipTestRule

Constructor Summary

Constructors

Constructor	Description
AbstractBlockingSnapshotTest()

Method Summary

Modifier and Type	Method	Description
void	aFailedBlockingSnapshotShouldNotCauseInitialSnapshotOnRestart()
void	anErrorDuringBlockingSnapshotShouldLeaveTheConnectorInAGoodState()
protected void	assertDdl(List<String> schemaChangesDdls)
protected void	assertRecordsFromSnapshotAndStreamingArePresent(int expectedRecords, AbstractConnectorTest.SourceRecords recordsByTopic)
protected void	assertRecordsWithValuesPresent(int expectedRecords, List<Integer> expectedValues, String topicName, AbstractConnectorTest.SourceRecords recordsByTopic)
protected void	assertStreamingRecordsArePresent(int expectedRecords, AbstractConnectorTest.SourceRecords recordsByTopic)
private void	assertStreamingTotalNumberOfCreateEventsSeen(Long expectedStreamingEvents)
protected abstract String	connector()
protected abstract JdbcConnection	databaseConnection()
protected abstract String	escapedTableDataCollectionId()
private Future<?>	executeAsync(Runnable operation)
void	executeBlockingSnapshot()
void	executeBlockingSnapshotWhileStreaming()
void	executeBlockingSnapshotWithAdditionalCondition()
void	executeBlockingSnapshotWithEscapedCollectionName()
protected int	expectedDdlsCount()
private static List<Integer>	getExpectedValues(Long totalSnapshotRecords)
private Long	getTotalSnapshotRecords(String table, String connector, String server, String task, String database)
private Long	getTotalStreamingCreateEventsSeen(String connector, String server, String task, String database)
protected Configuration.Builder	historizedMutableConfig(boolean signalTableOnly, boolean storeOnlyCapturedDdl)
protected int	insertMaxSleep()
protected void	insertRecords(int rowCount, int startingPkId)
private void	insertRecordsWithRandomSleep(int rowCount, int startingPkId, int maxSleep)
private void	insertRecordsWithRandomSleep(int rowCount, int startingPkId, int maxSleep, Runnable actionOnInsert)
private Runnable	insertTask()
protected abstract Configuration.Builder	mutableConfig(boolean signalTableOnly, boolean storeOnlyCapturedDdl)
void	readsSchemaOnlyForSignaledTables()
protected abstract String	server()
void	streamingMetricsResumeAfterBlockingSnapshot()
protected abstract String	tableName()
protected abstract String	topicName()
protected static void	waitForLogMessage(String message, Class<?> logEmitterClass)

Methods inherited from class io.debezium.pipeline.source.snapshot.incremental.AbstractSnapshotTest

alterTableAddColumnStatement, alterTableDropColumnStatement, buildAdditionalConditions, config, connectorClass, consumeMixedWithIncrementalSnapshot, consumeMixedWithIncrementalSnapshot, consumeMixedWithIncrementalSnapshot, consumeMixedWithIncrementalSnapshot, consumeMixedWithIncrementalSnapshot, consumeMixedWithIncrementalSnapshot, consumeMixedWithIncrementalSnapshot, consumeRecordsMixedWithIncrementalSnapshot, consumeRecordsMixedWithIncrementalSnapshot, database, getMaximumEnqueuedRecordCount, getRecordValue, pkFieldName, populate4PkTable, populateTable, populateTable, populateTable, populateTable, populateTables, populateTables, populateTableWithSpecificValue, sendAdHocSnapshotSignal, sendAdHocSnapshotSignalWithAdditionalConditionsWithSurrogateKey, sendAdHocSnapshotSignalWithAdditionalConditionsWithSurrogateKey, sendAdHocSnapshotSignalWithAdditionalConditionWithSurrogateKey, sendAdHocSnapshotSignalWithAdditionalConditionWithSurrogateKey, sendExecuteSnapshotFileSignal, sendExecuteSnapshotFileSignal, signalTableName, signalTableNameSanitized, startConnector, startConnector, startConnector, startConnector, startConnectorWithSnapshot, tableDataCollectionId, tableDataCollectionIds, tableNames, task, topicNames, valueFieldName, waitForCdcTransactionPropagation, waitForConnectorToStart

Methods inherited from class io.debezium.embedded.async.AbstractAsyncEngineConnectorTest

createEngine, createEngineBuilder, getSignaler

Methods inherited from class io.debezium.embedded.AbstractConnectorTest

assertBeginTransaction, assertConfigurationErrors, assertConfigurationErrors, assertConfigurationErrors, assertConnectorIsRunning, assertConnectorNotRunning, assertDelete, assertEndTransaction, assertEngineIsRunning, assertHasNoSourceQuery, assertInsert, assertKey, assertNoConfigurationErrors, assertNoRecordsToConsume, assertOffset, assertOffset, assertOnlyTransactionRecordsToConsume, assertRecordTransactionMetadata, assertSchemaMatchesStruct, assertSchemaMatchesStruct, assertSourceQuery, assertTombstone, assertTombstone, assertUpdate, assertValueField, cleanupTestFwkState, configValue, consumeAvailableRecords, consumeAvailableRecordsByTopic, consumeDmlRecordsByTopic, consumeDmlRecordsByTopic, consumeDmlRecordsByTopic, consumeRecord, consumeRecords, consumeRecords, consumeRecords, consumeRecordsButSkipUntil, consumeRecordsByTopic, consumeRecordsByTopic, consumeRecordsByTopic, consumeRecordsByTopicUntil, consumeRecordsUntil, debug, getConsumer, getSnapshotMetricsObjectName, getSnapshotMetricsObjectName, getSnapshotMetricsObjectName, getSnapshotMetricsObjectName, getStreamingMetricsObjectName, getStreamingMetricsObjectName, getStreamingMetricsObjectName, getStreamingMetricsObjectName, getStreamingMetricsObjectName, getStreamingMetricsObjectName, getStreamingNamespace, initializeConnectorTestFramework, isStreamingRunning, isStreamingRunning, isStreamingRunning, isStreamingRunning, isTransactionRecord, loggingCompletion, print, readLastCommittedOffset, readLastCommittedOffsets, setConsumeTimeout, skipAvroValidation, start, start, start, start, start, start, start, startAndConsumeTillEnd, startAndConsumeTillEnd, stopConnector, stopConnector, storeOffsets, validate, waitForAvailableRecords, waitForAvailableRecords, waitForConnectorShutdown, waitForEngineShutdown, waitForSnapshotToBeCompleted, waitForSnapshotToBeCompleted, waitForSnapshotWithCustomMetricsToBeCompleted, waitForSnapshotWithCustomMetricsToBeCompleted, waitForStreamingRunning, waitForStreamingRunning, waitForStreamingRunning, waitForStreamingWithCustomMetricsToStart, waitForStreamingWithCustomMetricsToStart, waitTimeForEngine, waitTimeForRecords, waitTimeForRecordsAfterNulls

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.debezium.util.Testing

once, once, resetBeforeEachTest, time, time

Field Details

signalingRecords

private int signalingRecords

ROW_COUNT

protected static final int ROW_COUNT

See Also:

• Constant Field Values

Constructor Details

AbstractBlockingSnapshotTest

public AbstractBlockingSnapshotTest()

Method Details

mutableConfig

protected abstract Configuration.Builder mutableConfig(boolean signalTableOnly,
 boolean storeOnlyCapturedDdl)

Specified by:

mutableConfig in class AbstractSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

databaseConnection

protected abstract JdbcConnection databaseConnection()

Specified by:

databaseConnection in class AbstractSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

topicName

protected abstract String topicName()

Specified by:

topicName in class AbstractSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

tableName

protected abstract String tableName()

Specified by:

tableName in class AbstractSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

escapedTableDataCollectionId

protected abstract String escapedTableDataCollectionId()

connector

protected abstract String connector()

Specified by:

connector in class AbstractSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

server

protected abstract String server()

Specified by:

server in class AbstractSnapshotTest<T extends org.apache.kafka.connect.source.SourceConnector>

historizedMutableConfig

protected Configuration.Builder historizedMutableConfig(boolean signalTableOnly,
 boolean storeOnlyCapturedDdl)

executeBlockingSnapshot

public void executeBlockingSnapshot()
                             throws Exception

Throws:

Exception

executeBlockingSnapshotWhileStreaming

public void executeBlockingSnapshotWhileStreaming()
                                           throws Exception

Throws:

Exception

executeBlockingSnapshotWithAdditionalCondition

public void executeBlockingSnapshotWithAdditionalCondition()
                                                    throws Exception

Throws:

Exception

streamingMetricsResumeAfterBlockingSnapshot

@FixFor("DBZ-8238")
public void streamingMetricsResumeAfterBlockingSnapshot()
                                                 throws Exception

Throws:

Exception

readsSchemaOnlyForSignaledTables

public void readsSchemaOnlyForSignaledTables()
                                      throws Exception

Throws:

Exception

executeBlockingSnapshotWithEscapedCollectionName

@FixFor("DBZ-7718")
public void executeBlockingSnapshotWithEscapedCollectionName()
                                                      throws Exception

Throws:

Exception

anErrorDuringBlockingSnapshotShouldLeaveTheConnectorInAGoodState

@FixFor("DBZ-8244")
public void anErrorDuringBlockingSnapshotShouldLeaveTheConnectorInAGoodState()
                                                                      throws Exception

Throws:

Exception

aFailedBlockingSnapshotShouldNotCauseInitialSnapshotOnRestart

@FixFor("DBZ-7903")
public void aFailedBlockingSnapshotShouldNotCauseInitialSnapshotOnRestart()
                                                                   throws Exception

Throws:

Exception

expectedDdlsCount

protected int expectedDdlsCount()

assertDdl

protected void assertDdl(List<String> schemaChangesDdls)

insertMaxSleep

protected int insertMaxSleep()

insertTask

private Runnable insertTask()

getTotalStreamingCreateEventsSeen

private Long getTotalStreamingCreateEventsSeen(String connector,
 String server,
 String task,
 String database)
                                        throws MalformedObjectNameException,
ReflectionException,
AttributeNotFoundException,
InstanceNotFoundException,
MBeanException

Throws:

MalformedObjectNameException

ReflectionException

AttributeNotFoundException

InstanceNotFoundException

MBeanException

getTotalSnapshotRecords

private Long getTotalSnapshotRecords(String table,
 String connector,
 String server,
 String task,
 String database)
                              throws MalformedObjectNameException,
ReflectionException,
AttributeNotFoundException,
InstanceNotFoundException,
MBeanException

Throws:

MalformedObjectNameException

ReflectionException

AttributeNotFoundException

InstanceNotFoundException

MBeanException

getExpectedValues

private static List<Integer> getExpectedValues(Long totalSnapshotRecords)

waitForLogMessage

protected static void waitForLogMessage(String message,
 Class<?> logEmitterClass)

assertStreamingTotalNumberOfCreateEventsSeen

private void assertStreamingTotalNumberOfCreateEventsSeen(Long expectedStreamingEvents)
                                                   throws ReflectionException,
MalformedObjectNameException,
AttributeNotFoundException,
InstanceNotFoundException,
MBeanException

Throws:

ReflectionException

MalformedObjectNameException

AttributeNotFoundException

InstanceNotFoundException

MBeanException

executeAsync

private Future<?> executeAsync(Runnable operation)

assertStreamingRecordsArePresent

protected void assertStreamingRecordsArePresent(int expectedRecords,
 AbstractConnectorTest.SourceRecords recordsByTopic)

assertRecordsFromSnapshotAndStreamingArePresent

protected void assertRecordsFromSnapshotAndStreamingArePresent(int expectedRecords,
 AbstractConnectorTest.SourceRecords recordsByTopic)
                                                        throws InterruptedException

Throws:

InterruptedException

assertRecordsWithValuesPresent

protected void assertRecordsWithValuesPresent(int expectedRecords,
 List<Integer> expectedValues,
 String topicName,
 AbstractConnectorTest.SourceRecords recordsByTopic)

insertRecords

protected void insertRecords(int rowCount,
 int startingPkId)
                      throws SQLException

Throws:

SQLException

insertRecordsWithRandomSleep

private void insertRecordsWithRandomSleep(int rowCount,
 int startingPkId,
 int maxSleep,
 Runnable actionOnInsert)
                                   throws SQLException

Throws:

SQLException

insertRecordsWithRandomSleep

private void insertRecordsWithRandomSleep(int rowCount,
 int startingPkId,
 int maxSleep)
                                   throws SQLException

Throws:

SQLException