DBZ-3161 Refactor/Rename all references of MonitoredTables to CapturedTables
This commit is contained in:
parent
062c1c3860
commit
f0f8d5d180
1
.gitignore
vendored
1
.gitignore
vendored
@ -28,3 +28,4 @@ generated-sources/
|
|||||||
|
|
||||||
/state/
|
/state/
|
||||||
bin/
|
bin/
|
||||||
|
gen/
|
||||||
|
@ -106,7 +106,7 @@ public MySqlDatabaseSchema(MySqlConnectorConfig connectorConfig, MySqlValueConve
|
|||||||
*
|
*
|
||||||
* @return the array with the table names
|
* @return the array with the table names
|
||||||
*/
|
*/
|
||||||
public String[] monitoredTablesAsStringArray() {
|
public String[] capturedTablesAsStringArray() {
|
||||||
final Collection<TableId> tables = tableIds();
|
final Collection<TableId> tables = tableIds();
|
||||||
String[] ret = new String[tables.size()];
|
String[] ret = new String[tables.size()];
|
||||||
int i = 0;
|
int i = 0;
|
||||||
@ -180,7 +180,7 @@ public void applySchemaChange(SchemaChangeEvent schemaChange) {
|
|||||||
// - all DDLs if configured
|
// - all DDLs if configured
|
||||||
// - or global SET variables
|
// - or global SET variables
|
||||||
// - or DDLs for monitored objects
|
// - or DDLs for monitored objects
|
||||||
if (!databaseHistory.storeOnlyMonitoredTables() || isGlobalSetVariableStatement(schemaChange.getDdl(), schemaChange.getDatabase())
|
if (!databaseHistory.storeOnlyCapturedTables() || isGlobalSetVariableStatement(schemaChange.getDdl(), schemaChange.getDatabase())
|
||||||
|| schemaChange.getTables().stream().map(Table::id).anyMatch(filters.dataCollectionFilter()::isIncluded)) {
|
|| schemaChange.getTables().stream().map(Table::id).anyMatch(filters.dataCollectionFilter()::isIncluded)) {
|
||||||
LOGGER.debug("Recorded DDL statements for database '{}': {}", schemaChange.getDatabase(), schemaChange.getDdl());
|
LOGGER.debug("Recorded DDL statements for database '{}': {}", schemaChange.getDatabase(), schemaChange.getDdl());
|
||||||
record(schemaChange, schemaChange.getTableChanges());
|
record(schemaChange, schemaChange.getTableChanges());
|
||||||
@ -219,7 +219,7 @@ private List<SchemaChangeEvent> parseDdl(String ddlStatements, String databaseNa
|
|||||||
}
|
}
|
||||||
|
|
||||||
// No need to send schema events or store DDL if no table has changed
|
// No need to send schema events or store DDL if no table has changed
|
||||||
if (!databaseHistory.storeOnlyMonitoredTables() || isGlobalSetVariableStatement(ddlStatements, databaseName) || ddlChanges.anyMatch(filters)) {
|
if (!databaseHistory.storeOnlyCapturedTables() || isGlobalSetVariableStatement(ddlStatements, databaseName) || ddlChanges.anyMatch(filters)) {
|
||||||
|
|
||||||
// We are supposed to _also_ record the schema changes as SourceRecords, but these need to be filtered
|
// We are supposed to _also_ record the schema changes as SourceRecords, but these need to be filtered
|
||||||
// by database. Unfortunately, the databaseName on the event might not be the same database as that
|
// by database. Unfortunately, the databaseName on the event might not be the same database as that
|
||||||
@ -316,8 +316,8 @@ public boolean historyExists() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean storeOnlyMonitoredTables() {
|
public boolean storeOnlyCapturedTables() {
|
||||||
return databaseHistory.storeOnlyMonitoredTables();
|
return databaseHistory.storeOnlyCapturedTables();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -236,7 +236,7 @@ protected void releaseDataSnapshotLocks(RelationalSnapshotContext snapshotContex
|
|||||||
for (Iterator<SchemaChangeEvent> i = schemaEvents.iterator(); i.hasNext();) {
|
for (Iterator<SchemaChangeEvent> i = schemaEvents.iterator(); i.hasNext();) {
|
||||||
final SchemaChangeEvent event = i.next();
|
final SchemaChangeEvent event = i.next();
|
||||||
|
|
||||||
if (databaseSchema.storeOnlyMonitoredTables() && event.getDatabase() != null && event.getDatabase().length() != 0
|
if (databaseSchema.storeOnlyCapturedTables() && event.getDatabase() != null && event.getDatabase().length() != 0
|
||||||
&& !connectorConfig.getTableFilters().databaseFilter().test(event.getDatabase())) {
|
&& !connectorConfig.getTableFilters().databaseFilter().test(event.getDatabase())) {
|
||||||
LOGGER.debug("Skipping schema event as it belongs to a non-captured database: '{}'", event);
|
LOGGER.debug("Skipping schema event as it belongs to a non-captured database: '{}'", event);
|
||||||
continue;
|
continue;
|
||||||
@ -315,7 +315,7 @@ protected void readTableStructure(ChangeEventSourceContext sourceContext, Relati
|
|||||||
delayedSchemaSnapshotTables = Collect.minus(snapshotContext.capturedSchemaTables, snapshotContext.capturedTables);
|
delayedSchemaSnapshotTables = Collect.minus(snapshotContext.capturedSchemaTables, snapshotContext.capturedTables);
|
||||||
LOGGER.info("Tables for delayed schema capture: {}", delayedSchemaSnapshotTables);
|
LOGGER.info("Tables for delayed schema capture: {}", delayedSchemaSnapshotTables);
|
||||||
}
|
}
|
||||||
if (databaseSchema.storeOnlyMonitoredTables()) {
|
if (databaseSchema.storeOnlyCapturedTables()) {
|
||||||
capturedSchemaTables = snapshotContext.capturedTables;
|
capturedSchemaTables = snapshotContext.capturedTables;
|
||||||
LOGGER.info("Only monitored tables schema should be captured, capturing: {}", capturedSchemaTables);
|
LOGGER.info("Only monitored tables schema should be captured, capturing: {}", capturedSchemaTables);
|
||||||
}
|
}
|
||||||
@ -592,7 +592,7 @@ protected void createSchemaChangeEventsForTables(ChangeEventSourceContext source
|
|||||||
throw new InterruptedException("Interrupted while processing event " + event);
|
throw new InterruptedException("Interrupted while processing event " + event);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (databaseSchema.storeOnlyMonitoredTables() && event.getDatabase() != null && event.getDatabase().length() != 0
|
if (databaseSchema.storeOnlyCapturedTables() && event.getDatabase() != null && event.getDatabase().length() != 0
|
||||||
&& !connectorConfig.getTableFilters().databaseFilter().test(event.getDatabase())) {
|
&& !connectorConfig.getTableFilters().databaseFilter().test(event.getDatabase())) {
|
||||||
LOGGER.debug("Skipping schema event as it belongs to a non-captured database: '{}'", event);
|
LOGGER.debug("Skipping schema event as it belongs to a non-captured database: '{}'", event);
|
||||||
continue;
|
continue;
|
||||||
|
@ -42,7 +42,12 @@ public void globalLockReleased() {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public String[] getMonitoredTables() {
|
public String[] getMonitoredTables() {
|
||||||
return schema.monitoredTablesAsStringArray();
|
return schema.capturedTablesAsStringArray();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String[] getCapturedTables() {
|
||||||
|
return schema.capturedTablesAsStringArray();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -154,7 +154,12 @@ public void setMilliSecondsBehindSource(long value) {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public String[] getMonitoredTables() {
|
public String[] getMonitoredTables() {
|
||||||
return schema.monitoredTablesAsStringArray();
|
return schema.capturedTablesAsStringArray();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String[] getCapturedTables() {
|
||||||
|
return schema.capturedTablesAsStringArray();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -872,7 +872,7 @@ protected void handleUpdateTableMetadata(Event event) {
|
|||||||
* {@link MySqlConnectorConfig#INCONSISTENT_SCHEMA_HANDLING_MODE} configuration.
|
* {@link MySqlConnectorConfig#INCONSISTENT_SCHEMA_HANDLING_MODE} configuration.
|
||||||
*/
|
*/
|
||||||
private void informAboutUnknownTableIfRequired(Event event, TableId tableId, String typeToLog) {
|
private void informAboutUnknownTableIfRequired(Event event, TableId tableId, String typeToLog) {
|
||||||
if (tableId != null && context.dbSchema().isTableMonitored(tableId)) {
|
if (tableId != null && context.dbSchema().isTableCaptured(tableId)) {
|
||||||
metrics.onErroneousEvent("source = " + tableId + ", event " + event);
|
metrics.onErroneousEvent("source = " + tableId + ", event " + event);
|
||||||
EventHeaderV4 eventHeader = event.getHeader();
|
EventHeaderV4 eventHeader = event.getHeader();
|
||||||
|
|
||||||
|
@ -152,8 +152,14 @@ public void setMilliSecondsBehindSource(long value) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@Deprecated
|
||||||
public String[] getMonitoredTables() {
|
public String[] getMonitoredTables() {
|
||||||
return schema.monitoredTablesAsStringArray();
|
return schema.capturedTablesAsStringArray();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String[] getCapturedTables() {
|
||||||
|
return schema.capturedTablesAsStringArray();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -79,7 +79,7 @@ public class MySqlSchema extends RelationalDatabaseSchema {
|
|||||||
private final DdlChanges ddlChanges;
|
private final DdlChanges ddlChanges;
|
||||||
private final HistoryRecordComparator historyComparator;
|
private final HistoryRecordComparator historyComparator;
|
||||||
private final boolean skipUnparseableDDL;
|
private final boolean skipUnparseableDDL;
|
||||||
private final boolean storeOnlyMonitoredTablesDdl;
|
private final boolean storeOnlyCapturedTablesDdl;
|
||||||
private boolean recoveredTables;
|
private boolean recoveredTables;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -122,7 +122,7 @@ public MySqlSchema(MySqlConnectorConfig configuration,
|
|||||||
.with(KafkaDatabaseHistory.INTERNAL_CONNECTOR_ID, configuration.getLogicalName())
|
.with(KafkaDatabaseHistory.INTERNAL_CONNECTOR_ID, configuration.getLogicalName())
|
||||||
.build();
|
.build();
|
||||||
this.skipUnparseableDDL = dbHistoryConfig.getBoolean(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS);
|
this.skipUnparseableDDL = dbHistoryConfig.getBoolean(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS);
|
||||||
this.storeOnlyMonitoredTablesDdl = dbHistoryConfig.getBoolean(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL);
|
this.storeOnlyCapturedTablesDdl = dbHistoryConfig.getBoolean(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL);
|
||||||
|
|
||||||
this.ddlParser = new MySqlAntlrDdlParser(getValueConverters(configuration), getTableFilter());
|
this.ddlParser = new MySqlAntlrDdlParser(getValueConverters(configuration), getTableFilter());
|
||||||
this.ddlChanges = this.ddlParser.getDdlChanges();
|
this.ddlChanges = this.ddlParser.getDdlChanges();
|
||||||
@ -202,7 +202,7 @@ public Filters filters() {
|
|||||||
*
|
*
|
||||||
* @return the array with the table names
|
* @return the array with the table names
|
||||||
*/
|
*/
|
||||||
public String[] monitoredTablesAsStringArray() {
|
public String[] capturedTablesAsStringArray() {
|
||||||
final Collection<TableId> tables = tableIds();
|
final Collection<TableId> tables = tableIds();
|
||||||
String[] ret = new String[tables.size()];
|
String[] ret = new String[tables.size()];
|
||||||
int i = 0;
|
int i = 0;
|
||||||
@ -218,7 +218,7 @@ public String[] monitoredTablesAsStringArray() {
|
|||||||
* @param id the fully-qualified table identifier; may be null
|
* @param id the fully-qualified table identifier; may be null
|
||||||
* @return true if events from the table are captured
|
* @return true if events from the table are captured
|
||||||
*/
|
*/
|
||||||
public boolean isTableMonitored(TableId id) {
|
public boolean isTableCaptured(TableId id) {
|
||||||
return filters.tableFilter().test(id);
|
return filters.tableFilter().test(id);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -338,7 +338,7 @@ public boolean applyDdl(SourceInfo source, String databaseName, String ddlStatem
|
|||||||
}
|
}
|
||||||
changes = tables().drainChanges();
|
changes = tables().drainChanges();
|
||||||
// No need to send schema events or store DDL if no table has changed
|
// No need to send schema events or store DDL if no table has changed
|
||||||
if (!storeOnlyMonitoredTablesDdl || ddlChanges.anyMatch(filters.databaseFilter(), filters.tableFilter())) {
|
if (!storeOnlyCapturedTablesDdl || ddlChanges.anyMatch(filters.databaseFilter(), filters.tableFilter())) {
|
||||||
if (statementConsumer != null) {
|
if (statementConsumer != null) {
|
||||||
|
|
||||||
// We are supposed to _also_ record the schema changes as SourceRecords, but these need to be filtered
|
// We are supposed to _also_ record the schema changes as SourceRecords, but these need to be filtered
|
||||||
@ -377,7 +377,7 @@ else if (filters.databaseFilter().test(databaseName) || databaseName == null ||
|
|||||||
// - all DDLs if configured
|
// - all DDLs if configured
|
||||||
// - or global SET variables
|
// - or global SET variables
|
||||||
// - or DDLs for monitored objects
|
// - or DDLs for monitored objects
|
||||||
if (!storeOnlyMonitoredTablesDdl || isGlobalSetVariableStatement(ddlStatements, databaseName) || changes.stream().anyMatch(filters().tableFilter()::test)) {
|
if (!storeOnlyCapturedTablesDdl || isGlobalSetVariableStatement(ddlStatements, databaseName) || changes.stream().anyMatch(filters().tableFilter()::test)) {
|
||||||
dbHistory.record(source.partition(), source.offset(), databaseName, ddlStatements);
|
dbHistory.record(source.partition(), source.offset(), databaseName, ddlStatements);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -403,10 +403,10 @@ public boolean isGlobalSetVariableStatement(String ddl, String databaseName) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @return true if only monitored tables should be stored in database history, false if all tables should be stored
|
* @return true if only captured tables should be stored in database history, false if all tables should be stored
|
||||||
*/
|
*/
|
||||||
public boolean isStoreOnlyMonitoredTablesDdl() {
|
public boolean isStoreOnlyCapturedTablesDdl() {
|
||||||
return storeOnlyMonitoredTablesDdl;
|
return storeOnlyCapturedTablesDdl;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -13,5 +13,9 @@
|
|||||||
*/
|
*/
|
||||||
public interface ReaderMetricsMXBean {
|
public interface ReaderMetricsMXBean {
|
||||||
|
|
||||||
|
@Deprecated
|
||||||
String[] getMonitoredTables();
|
String[] getMonitoredTables();
|
||||||
|
|
||||||
|
String[] getCapturedTables();
|
||||||
|
|
||||||
}
|
}
|
||||||
|
@ -800,7 +800,7 @@ private boolean shouldRecordTableSchema(MySqlSchema schema, Filters filters, Tab
|
|||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
return filters.tableFilter().test(id) || !schema.isStoreOnlyMonitoredTablesDdl();
|
return filters.tableFilter().test(id) || !schema.isStoreOnlyCapturedTablesDdl();
|
||||||
}
|
}
|
||||||
|
|
||||||
protected void readBinlogPosition(int step, SourceInfo source, JdbcConnection mysql, AtomicReference<String> sql) throws SQLException {
|
protected void readBinlogPosition(int step, SourceInfo source, JdbcConnection mysql, AtomicReference<String> sql) throws SQLException {
|
||||||
|
@ -40,7 +40,12 @@ public void globalLockReleased() {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public String[] getMonitoredTables() {
|
public String[] getMonitoredTables() {
|
||||||
return schema.monitoredTablesAsStringArray();
|
return schema.capturedTablesAsStringArray();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String[] getCapturedTables() {
|
||||||
|
return schema.capturedTablesAsStringArray();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -775,7 +775,7 @@ public void shouldUseOverriddenSelectStatementDuringSnapshotting() throws SQLExc
|
|||||||
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE, DATABASE.getDatabaseName() + ".products")
|
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE, DATABASE.getDatabaseName() + ".products")
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE + "." + DATABASE.getDatabaseName() + ".products",
|
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE + "." + DATABASE.getDatabaseName() + ".products",
|
||||||
String.format("SELECT * from %s.products where id>=108 order by id", DATABASE.getDatabaseName()))
|
String.format("SELECT * from %s.products where id>=108 order by id", DATABASE.getDatabaseName()))
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.with(MySqlConnectorConfig.DATABASE_HISTORY, FileDatabaseHistory.class)
|
.with(MySqlConnectorConfig.DATABASE_HISTORY, FileDatabaseHistory.class)
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
||||||
@ -823,7 +823,7 @@ public void shouldUseMultipleOverriddenSelectStatementsDuringSnapshotting() thro
|
|||||||
.with(MySqlConnectorConfig.POLL_INTERVAL_MS, 10)
|
.with(MySqlConnectorConfig.POLL_INTERVAL_MS, 10)
|
||||||
.with(MySqlConnectorConfig.DATABASE_INCLUDE_LIST, DATABASE.getDatabaseName())
|
.with(MySqlConnectorConfig.DATABASE_INCLUDE_LIST, DATABASE.getDatabaseName())
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, tables)
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, tables)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE, tables)
|
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE, tables)
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE + "." + DATABASE.getDatabaseName() + ".products",
|
.with(MySqlConnectorConfig.SNAPSHOT_SELECT_STATEMENT_OVERRIDES_BY_TABLE + "." + DATABASE.getDatabaseName() + ".products",
|
||||||
String.format("SELECT * from %s.products where id>=108 order by id", DATABASE.getDatabaseName()))
|
String.format("SELECT * from %s.products where id>=108 order by id", DATABASE.getDatabaseName()))
|
||||||
@ -867,7 +867,7 @@ public void shouldIgnoreAlterTableForNonCapturedTablesNotStoredInHistory() throw
|
|||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, tables)
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, tables)
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.build();
|
.build();
|
||||||
|
|
||||||
// Start the connector ...
|
// Start the connector ...
|
||||||
@ -896,14 +896,14 @@ public void shouldIgnoreAlterTableForNonCapturedTablesNotStoredInHistory() throw
|
|||||||
|
|
||||||
@Test
|
@Test
|
||||||
@FixFor("DBZ-1201")
|
@FixFor("DBZ-1201")
|
||||||
public void shouldSaveSetCharacterSetWhenStoringOnlyMonitoredTables() throws SQLException, InterruptedException {
|
public void shouldSaveSetCharacterSetWhenStoringOnlyCapturededTables() throws SQLException, InterruptedException {
|
||||||
Testing.Files.delete(DB_HISTORY_PATH);
|
Testing.Files.delete(DB_HISTORY_PATH);
|
||||||
|
|
||||||
config = DATABASE.defaultConfig()
|
config = DATABASE.defaultConfig()
|
||||||
.with(MySqlConnectorConfig.DATABASE_INCLUDE_LIST, "no_" + DATABASE.getDatabaseName())
|
.with(MySqlConnectorConfig.DATABASE_INCLUDE_LIST, "no_" + DATABASE.getDatabaseName())
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.build();
|
.build();
|
||||||
|
|
||||||
// Start the connector ...
|
// Start the connector ...
|
||||||
@ -1016,7 +1016,7 @@ public void shouldIgnoreCreateIndexForNonCapturedTablesNotStoredInHistory() thro
|
|||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, tables)
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, tables)
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.build();
|
.build();
|
||||||
|
|
||||||
try (MySqlTestConnection db = MySqlTestConnection.forTestDatabase(DATABASE.getDatabaseName());) {
|
try (MySqlTestConnection db = MySqlTestConnection.forTestDatabase(DATABASE.getDatabaseName());) {
|
||||||
|
@ -850,7 +850,7 @@ public void shouldConsumeDatesCorrectlyWhenClientTimezonePrecedesServerTimezoneU
|
|||||||
// Use the DB configuration to define the connector's configuration ...
|
// Use the DB configuration to define the connector's configuration ...
|
||||||
config = DATABASE.defaultConfig()
|
config = DATABASE.defaultConfig()
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("dbz_85_fractest"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("dbz_85_fractest"))
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.build();
|
.build();
|
||||||
// Start the connector ...
|
// Start the connector ...
|
||||||
start(MySqlConnector.class, config);
|
start(MySqlConnector.class, config);
|
||||||
@ -1011,7 +1011,7 @@ public void shouldConsumeDecimalAsStringFromSnapshot() throws SQLException, Inte
|
|||||||
config = DATABASE.defaultConfig()
|
config = DATABASE.defaultConfig()
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("dbz_147_decimalvalues"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("dbz_147_decimalvalues"))
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.with(MySqlConnectorConfig.DECIMAL_HANDLING_MODE, DecimalHandlingMode.STRING)
|
.with(MySqlConnectorConfig.DECIMAL_HANDLING_MODE, DecimalHandlingMode.STRING)
|
||||||
.build();
|
.build();
|
||||||
// Start the connector ...
|
// Start the connector ...
|
||||||
|
@ -72,7 +72,7 @@ public void testLifecycle() throws Exception {
|
|||||||
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
||||||
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, Boolean.TRUE)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, Boolean.TRUE)
|
||||||
.build());
|
.build());
|
||||||
|
|
||||||
assertConnectorIsRunning();
|
assertConnectorIsRunning();
|
||||||
@ -120,7 +120,7 @@ public void testSnapshotOnlyMetrics() throws Exception {
|
|||||||
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
||||||
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, Boolean.TRUE)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, Boolean.TRUE)
|
||||||
.build());
|
.build());
|
||||||
|
|
||||||
assertSnapshotMetrics();
|
assertSnapshotMetrics();
|
||||||
@ -151,7 +151,7 @@ public void testSnapshotAndStreamingMetrics() throws Exception {
|
|||||||
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
||||||
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, Boolean.TRUE)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, Boolean.TRUE)
|
||||||
.build());
|
.build());
|
||||||
|
|
||||||
assertSnapshotMetrics();
|
assertSnapshotMetrics();
|
||||||
@ -168,7 +168,7 @@ public void testStreamingOnlyMetrics() throws Exception {
|
|||||||
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
.with(FileDatabaseHistory.FILE_PATH, DB_HISTORY_PATH)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("simple"))
|
||||||
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
.with(MySqlConnectorConfig.TABLES_IGNORE_BUILTIN, Boolean.TRUE)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, Boolean.TRUE)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, Boolean.TRUE)
|
||||||
.build());
|
.build());
|
||||||
|
|
||||||
// CREATE DATABASE, CREATE TABLE, and 2 INSERT
|
// CREATE DATABASE, CREATE TABLE, and 2 INSERT
|
||||||
|
@ -149,7 +149,7 @@ public void shouldWarnOnInvalidMigrateTable() throws SQLException, InterruptedEx
|
|||||||
config = DATABASE.defaultConfig()
|
config = DATABASE.defaultConfig()
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, false)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, false)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("monitored"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("monitored"))
|
||||||
.build();
|
.build();
|
||||||
|
|
||||||
|
@ -646,7 +646,7 @@ public void dateAndTimeTest() throws InterruptedException {
|
|||||||
config = DATABASE.defaultConfig()
|
config = DATABASE.defaultConfig()
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("DATE_TIME_TABLE"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("DATE_TIME_TABLE"))
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.build();
|
.build();
|
||||||
start(MySqlConnector.class, config);
|
start(MySqlConnector.class, config);
|
||||||
|
|
||||||
@ -715,7 +715,7 @@ public void timeTypeWithConnectMode() throws Exception {
|
|||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("DATE_TIME_TABLE"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("DATE_TIME_TABLE"))
|
||||||
.with(MySqlConnectorConfig.TIME_PRECISION_MODE, TemporalPrecisionMode.CONNECT)
|
.with(MySqlConnectorConfig.TIME_PRECISION_MODE, TemporalPrecisionMode.CONNECT)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.build();
|
.build();
|
||||||
start(MySqlConnector.class, config);
|
start(MySqlConnector.class, config);
|
||||||
|
|
||||||
@ -858,7 +858,7 @@ public void alterDateAndTimeTest() throws Exception {
|
|||||||
config = DATABASE.defaultConfig()
|
config = DATABASE.defaultConfig()
|
||||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("ALTER_DATE_TIME"))
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("ALTER_DATE_TIME"))
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, false)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, false)
|
||||||
.build();
|
.build();
|
||||||
start(MySqlConnector.class, config);
|
start(MySqlConnector.class, config);
|
||||||
|
@ -108,7 +108,7 @@ public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLock() throws Excep
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockAndStoreOnlyMonitoredTables() throws Exception {
|
public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockAndStoreOnlyCapturedTables() throws Exception {
|
||||||
snapshotOfSingleDatabase(false, true, true);
|
snapshotOfSingleDatabase(false, true, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -123,11 +123,11 @@ public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockNoData() throws
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockAndStoreOnlyMonitoredTablesNoData() throws Exception {
|
public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockAndStoreOnlyCapturedTablesNoData() throws Exception {
|
||||||
snapshotOfSingleDatabase(false, true, false);
|
snapshotOfSingleDatabase(false, true, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyMonitoredTables, boolean data) throws Exception {
|
private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyCapturedTables, boolean data) throws Exception {
|
||||||
final LogInterceptor logInterceptor = new LogInterceptor();
|
final LogInterceptor logInterceptor = new LogInterceptor();
|
||||||
|
|
||||||
final Builder builder = simpleConfig()
|
final Builder builder = simpleConfig()
|
||||||
@ -138,7 +138,7 @@ private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyMo
|
|||||||
.with(MySqlConnectorConfig.PASSWORD, "cloudpass")
|
.with(MySqlConnectorConfig.PASSWORD, "cloudpass")
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(MySqlConnectorConfig.TEST_DISABLE_GLOBAL_LOCKING, "true")
|
.with(MySqlConnectorConfig.TEST_DISABLE_GLOBAL_LOCKING, "true")
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, storeOnlyMonitoredTables);
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, storeOnlyCapturedTables);
|
||||||
}
|
}
|
||||||
if (!data) {
|
if (!data) {
|
||||||
builder.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY);
|
builder.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY);
|
||||||
@ -153,7 +153,7 @@ private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyMo
|
|||||||
// Testing.Print.enable();
|
// Testing.Print.enable();
|
||||||
KeyValueStore store = KeyValueStore.createForTopicsBeginningWith(DATABASE.getServerName() + ".");
|
KeyValueStore store = KeyValueStore.createForTopicsBeginningWith(DATABASE.getServerName() + ".");
|
||||||
SchemaChangeHistory schemaChanges = new SchemaChangeHistory(DATABASE.getServerName());
|
SchemaChangeHistory schemaChanges = new SchemaChangeHistory(DATABASE.getServerName());
|
||||||
final int schemaEventsCount = !useGlobalLock ? (storeOnlyMonitoredTables ? 8 : 14) : 0;
|
final int schemaEventsCount = !useGlobalLock ? (storeOnlyCapturedTables ? 8 : 14) : 0;
|
||||||
SourceRecords sourceRecords = consumeRecordsByTopic(schemaEventsCount + (data ? 9 + 4 : 0));
|
SourceRecords sourceRecords = consumeRecordsByTopic(schemaEventsCount + (data ? 9 + 4 : 0));
|
||||||
for (Iterator<SourceRecord> i = sourceRecords.allRecordsInOrder().iterator(); i.hasNext();) {
|
for (Iterator<SourceRecord> i = sourceRecords.allRecordsInOrder().iterator(); i.hasNext();) {
|
||||||
final SourceRecord record = i.next();
|
final SourceRecord record = i.next();
|
||||||
|
@ -525,7 +525,7 @@ public void shouldAcceptTls12() throws Exception {
|
|||||||
|
|
||||||
private void inconsistentSchema(EventProcessingFailureHandlingMode mode) throws InterruptedException, SQLException {
|
private void inconsistentSchema(EventProcessingFailureHandlingMode mode) throws InterruptedException, SQLException {
|
||||||
Configuration.Builder builder = simpleConfig()
|
Configuration.Builder builder = simpleConfig()
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, true)
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, true)
|
||||||
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("orders"));
|
.with(MySqlConnectorConfig.TABLE_INCLUDE_LIST, DATABASE.qualifiedTableName("orders"));
|
||||||
|
|
||||||
if (mode == null) {
|
if (mode == null) {
|
||||||
|
@ -118,18 +118,18 @@ public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLock() throws Excep
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockAndStoreOnlyMonitoredTables() throws Exception {
|
public void shouldCreateSnapshotOfSingleDatabaseWithoutGlobalLockAndStoreOnlyCapturedTables() throws Exception {
|
||||||
snapshotOfSingleDatabase(false, true);
|
snapshotOfSingleDatabase(false, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyMonitoredTables) throws Exception {
|
private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyCapturedTables) throws Exception {
|
||||||
final Builder builder = simpleConfig();
|
final Builder builder = simpleConfig();
|
||||||
if (!useGlobalLock) {
|
if (!useGlobalLock) {
|
||||||
builder
|
builder
|
||||||
.with(MySqlConnectorConfig.USER, "cloud")
|
.with(MySqlConnectorConfig.USER, "cloud")
|
||||||
.with(MySqlConnectorConfig.PASSWORD, "cloudpass")
|
.with(MySqlConnectorConfig.PASSWORD, "cloudpass")
|
||||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, true)
|
||||||
.with(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL, storeOnlyMonitoredTables);
|
.with(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL, storeOnlyCapturedTables);
|
||||||
}
|
}
|
||||||
config = builder.build();
|
config = builder.build();
|
||||||
context = new MySqlTaskContext(config, new Filters.Builder(config).build());
|
context = new MySqlTaskContext(config, new Filters.Builder(config).build());
|
||||||
|
@ -2551,7 +2551,7 @@ public void initializeStorage() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean storeOnlyMonitoredTables() {
|
public boolean storeOnlyCapturedTables() {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -276,7 +276,7 @@ public Optional<DataCollectionSchema> ignoreMissingSchema(T dataCollectionId, Ch
|
|||||||
|
|
||||||
public void dispatchSchemaChangeEvent(T dataCollectionId, SchemaChangeEventEmitter schemaChangeEventEmitter) throws InterruptedException {
|
public void dispatchSchemaChangeEvent(T dataCollectionId, SchemaChangeEventEmitter schemaChangeEventEmitter) throws InterruptedException {
|
||||||
if (dataCollectionId != null && !filter.isIncluded(dataCollectionId)) {
|
if (dataCollectionId != null && !filter.isIncluded(dataCollectionId)) {
|
||||||
if (historizedSchema == null || historizedSchema.storeOnlyMonitoredTables()) {
|
if (historizedSchema == null || historizedSchema.storeOnlyCapturedTables()) {
|
||||||
LOGGER.trace("Filtering schema change event for {}", dataCollectionId);
|
LOGGER.trace("Filtering schema change event for {}", dataCollectionId);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@ -298,7 +298,7 @@ public void dispatchSchemaChangeEvent(Collection<T> dataCollectionIds, SchemaCha
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (!anyNonfilteredEvent) {
|
if (!anyNonfilteredEvent) {
|
||||||
if (historizedSchema == null || historizedSchema.storeOnlyMonitoredTables()) {
|
if (historizedSchema == null || historizedSchema.storeOnlyCapturedTables()) {
|
||||||
LOGGER.trace("Filtering schema change event for {}", dataCollectionIds);
|
LOGGER.trace("Filtering schema change event for {}", dataCollectionIds);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
@ -94,7 +94,7 @@ public boolean tableInformationComplete() {
|
|||||||
return recoveredTables;
|
return recoveredTables;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean storeOnlyMonitoredTables() {
|
public boolean storeOnlyCapturedTables() {
|
||||||
return databaseHistory.storeOnlyMonitoredTables();
|
return databaseHistory.storeOnlyCapturedTables();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -49,7 +49,7 @@ public abstract class AbstractDatabaseHistory implements DatabaseHistory {
|
|||||||
protected Configuration config;
|
protected Configuration config;
|
||||||
private HistoryRecordComparator comparator = HistoryRecordComparator.INSTANCE;
|
private HistoryRecordComparator comparator = HistoryRecordComparator.INSTANCE;
|
||||||
private boolean skipUnparseableDDL;
|
private boolean skipUnparseableDDL;
|
||||||
private boolean storeOnlyMonitoredTablesDdl;
|
private boolean storeOnlyCapturedTablesDdl;
|
||||||
private Function<String, Optional<Pattern>> ddlFilter = (x -> Optional.empty());
|
private Function<String, Optional<Pattern>> ddlFilter = (x -> Optional.empty());
|
||||||
private DatabaseHistoryListener listener = DatabaseHistoryListener.NOOP;
|
private DatabaseHistoryListener listener = DatabaseHistoryListener.NOOP;
|
||||||
private boolean useCatalogBeforeSchema;
|
private boolean useCatalogBeforeSchema;
|
||||||
@ -64,7 +64,7 @@ public void configure(Configuration config, HistoryRecordComparator comparator,
|
|||||||
this.config = config;
|
this.config = config;
|
||||||
this.comparator = comparator != null ? comparator : HistoryRecordComparator.INSTANCE;
|
this.comparator = comparator != null ? comparator : HistoryRecordComparator.INSTANCE;
|
||||||
this.skipUnparseableDDL = config.getBoolean(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS);
|
this.skipUnparseableDDL = config.getBoolean(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS);
|
||||||
this.storeOnlyMonitoredTablesDdl = config.getBoolean(DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL);
|
this.storeOnlyCapturedTablesDdl = config.getBoolean(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL);
|
||||||
|
|
||||||
final String ddlFilter = config.getString(DatabaseHistory.DDL_FILTER);
|
final String ddlFilter = config.getString(DatabaseHistory.DDL_FILTER);
|
||||||
this.ddlFilter = (ddlFilter != null) ? Predicates.matchedBy(ddlFilter) : this.ddlFilter;
|
this.ddlFilter = (ddlFilter != null) ? Predicates.matchedBy(ddlFilter) : this.ddlFilter;
|
||||||
@ -165,8 +165,8 @@ public void initializeStorage() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean storeOnlyMonitoredTables() {
|
public boolean storeOnlyCapturedTables() {
|
||||||
return storeOnlyMonitoredTablesDdl;
|
return storeOnlyCapturedTablesDdl;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -158,7 +158,7 @@ public interface DatabaseHistory {
|
|||||||
*/
|
*/
|
||||||
void initializeStorage();
|
void initializeStorage();
|
||||||
|
|
||||||
boolean storeOnlyMonitoredTables();
|
boolean storeOnlyCapturedTables();
|
||||||
|
|
||||||
boolean skipUnparseableDdlStatements();
|
boolean skipUnparseableDdlStatements();
|
||||||
|
|
||||||
|
@ -36,7 +36,7 @@ public static interface SchemaChangeEventConsumer {
|
|||||||
|
|
||||||
void initializeStorage();
|
void initializeStorage();
|
||||||
|
|
||||||
default boolean storeOnlyMonitoredTables() {
|
default boolean storeOnlyCapturedTables() {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user