MySqlTaskContext extends CdcSourceTaskContext { public Predicate<String> ddlFilter() { return ddlFilter; } MySqlTaskContext(Configuration config, Filters filters);  MySqlTaskContext(Configuration config, Filters filters, Map<String, ?> restartOffset);  MySqlTaskContext(Configuration config, Filters filters, Boolean tableIdCaseInsensitive, Map<String, ?> restartOffset); Configuration config(); MySqlConnectorConfig getConnectorConfig(); MySqlJdbcContext getConnectionContext(); String connectorName(); TopicSelector<TableId> topicSelector(); SourceInfo source(); MySqlSchema dbSchema(); RecordMakers makeRecord(); Predicate<String> gtidSourceFilter(); void initializeHistory(); void loadHistory(SourceInfo startingPoint); boolean historyExists(); void initializeHistoryStorage(); long serverId(); long rowCountForLargeTable(); int bufferSizeForBinlogReader(); boolean includeSchemaChangeRecords(); boolean includeSqlQuery(); boolean isSnapshotAllowedWhenNeeded(); boolean isSnapshotNeverAllowed(); boolean isInitialSnapshotOnly(); boolean isSchemaOnlySnapshot(); boolean isSchemaOnlyRecoverySnapshot(); void start(); void shutdown(); void temporaryLoggingContext(String contextName, Runnable operation); GtidSet filterGtidSet(GtidSet availableServerGtidSet, GtidSet purgedServerGtid); Predicate<String> ddlFilter(); boolean isTableIdCaseInsensitive();  }
@Test public void shouldFilterInternalDmlStatementsUsingDefaultFilter() throws Exception { config = simpleConfig().build(); context = new MySqlTaskContext(config, new Filters.Builder(config).build(), false, null); assertThat(context.ddlFilter().test("INSERT INTO mysql.rds_heartbeat2(name) values ('innodb_txn_key') ON DUPLICATE KEY UPDATE value = 'v'")).isTrue(); assertThat(context.ddlFilter().test("INSERT INTO mysql.rds_sysinfo(name, value) values ('innodb_txn_key','Sat Jun 13 06:26:02 UTC 2020')")).isTrue(); assertThat(context.ddlFilter().test("INSERT INTO mysql.rds_monitor(name, value) values ('innodb_txn_key','Sat Jun 13 06:26:02 UTC 2020')")).isTrue(); assertThat(context.ddlFilter().test("INSERT INTO mysql.rds_monitor(name) values ('innodb_txn_key') ON DUPLICATE KEY UPDATE value = 'v'")).isTrue(); assertThat(context.ddlFilter().test("DELETE FROM mysql.rds_sysinfo")).isTrue(); assertThat(context.ddlFilter().test("DELETE FROM mysql.rds_monitor;")).isTrue(); assertThat(context.ddlFilter().test("FLUSH RELAY LOGS;")).isTrue(); assertThat(context.ddlFilter().test("SAVEPOINT x")).isTrue(); assertThat(context.ddlFilter().test("INSERT INTO mysql.rds_heartbeat2(name) values ('innodb_txn_key')")).isFalse(); assertThat(context.ddlFilter().test("SAVEPOINT;")).isFalse(); }