incrementalEventSource =
getIncrementalSnapshotChangeEventSource();
if (incrementalEventSource != null) {
- incrementalEventSource.processSchemaChange(partition, dataCollectionId);
+ incrementalEventSource.processSchemaChange(partition, offsetContext, dataCollectionId);
}
}
@@ -200,7 +202,8 @@ private Struct schemaChangeRecordValue(SchemaChangeEvent event) throws IOExcepti
event.getDatabase(),
event.getSchema(),
event.getDdl(),
- event.getTableChanges());
+ event.getTableChanges(),
+ event.getTimestamp());
String historyStr = DOCUMENT_WRITER.write(historyRecord.document());
Struct value = new Struct(schemaChangeValueSchema);
@@ -217,7 +220,7 @@ public void schemaChangeEvent(SchemaChangeEvent event) throws InterruptedExcepti
historizedSchema.applySchemaChange(event);
if (connectorConfig.isSchemaChangesHistoryEnabled()) {
try {
- final String topicName = topicSelector.getPrimaryTopic();
+ final String topicName = topic;
final Integer partition = 0;
final Struct key = schemaChangeRecordKey(event);
final Struct value = schemaChangeRecordValue(event);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/EmbeddedFlinkDatabaseHistory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/EmbeddedFlinkDatabaseHistory.java
index 32e5d2198b1..aa992711eb7 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/EmbeddedFlinkDatabaseHistory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/EmbeddedFlinkDatabaseHistory.java
@@ -23,14 +23,15 @@
import io.debezium.relational.TableId;
import io.debezium.relational.Tables;
import io.debezium.relational.ddl.DdlParser;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryException;
-import io.debezium.relational.history.DatabaseHistoryListener;
import io.debezium.relational.history.HistoryRecord;
import io.debezium.relational.history.HistoryRecordComparator;
+import io.debezium.relational.history.SchemaHistory;
+import io.debezium.relational.history.SchemaHistoryException;
+import io.debezium.relational.history.SchemaHistoryListener;
import io.debezium.relational.history.TableChanges;
import io.debezium.relational.history.TableChanges.TableChange;
+import java.time.Instant;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -39,19 +40,20 @@
import java.util.concurrent.ConcurrentMap;
/**
- * A {@link DatabaseHistory} implementation which store the latest table schema in Flink state.
+ * A {@link SchemaHistory} implementation which store the latest table schema in Flink state.
*
*
It stores/recovers history using data offered by {@link SourceSplitState}.
*/
-public class EmbeddedFlinkDatabaseHistory implements DatabaseHistory {
+public class EmbeddedFlinkDatabaseHistory implements SchemaHistory {
- public static final String DATABASE_HISTORY_INSTANCE_NAME = "database.history.instance.name";
+ public static final String DATABASE_HISTORY_INSTANCE_NAME =
+ "schema.history.internal.instance.name";
public static final ConcurrentMap> TABLE_SCHEMAS =
new ConcurrentHashMap<>();
private Map tableSchemas;
- private DatabaseHistoryListener listener;
+ private SchemaHistoryListener listener;
private boolean storeOnlyMonitoredTablesDdl;
private boolean skipUnparseableDDL;
@@ -59,10 +61,10 @@ public class EmbeddedFlinkDatabaseHistory implements DatabaseHistory {
public void configure(
Configuration config,
HistoryRecordComparator comparator,
- DatabaseHistoryListener listener,
+ SchemaHistoryListener listener,
boolean useCatalogBeforeSchema) {
this.listener = listener;
- this.storeOnlyMonitoredTablesDdl = config.getBoolean(STORE_ONLY_MONITORED_TABLES_DDL);
+ this.storeOnlyMonitoredTablesDdl = config.getBoolean(STORE_ONLY_CAPTURED_TABLES_DDL);
this.skipUnparseableDDL = config.getBoolean(SKIP_UNPARSEABLE_DDL_STATEMENTS);
// recover
@@ -74,14 +76,19 @@ public void configure(
}
@Override
- public void start() {
- listener.started();
- }
+ // Debezium 2.0 wires SchemaHistoryMetrics in as the schema history listener. Its
+ // started() callback registers a JMX MBean whose name is built only from the connector
+ // context and the topic prefix, so every parallel subtask sharing a TaskManager JVM asks
+ // for the very same name. Debezium answers a name clash by sleeping 5 seconds and
+ // retrying, twelve times, so every reader but the first stalls for up to a minute each
+ // time it opens a split. Flink CDC publishes its own metrics and never reads these
+ // MBeans, and Debezium 1.9 did not register them either, so skip the registration.
+ public void start() {}
@Override
public void record(
Map source, Map position, String databaseName, String ddl)
- throws DatabaseHistoryException {
+ throws SchemaHistoryException {
throw new UnsupportedOperationException("should not call here, error");
}
@@ -92,10 +99,12 @@ public void record(
String databaseName,
String schemaName,
String ddl,
- TableChanges changes)
- throws DatabaseHistoryException {
+ TableChanges changes,
+ Instant timestamp)
+ throws SchemaHistoryException {
final HistoryRecord record =
- new HistoryRecord(source, position, databaseName, schemaName, ddl, changes);
+ new HistoryRecord(
+ source, position, databaseName, schemaName, ddl, changes, timestamp);
listener.onChangeApplied(record);
}
@@ -135,12 +144,12 @@ public void initializeStorage() {
// do nothing
}
- @Override
+ // Debezium 2.2 moved storeOnlyCapturedTables()/skipUnparseableDdlStatements() from the
+ // SchemaHistory interface to HistorizedDatabaseSchema; kept here as plain helpers.
public boolean storeOnlyCapturedTables() {
return storeOnlyMonitoredTablesDdl;
}
- @Override
public boolean skipUnparseableDdlStatements() {
return skipUnparseableDDL;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/wartermark/WatermarkEvent.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/wartermark/WatermarkEvent.java
index e57335ab420..b48ae7b9b5c 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/wartermark/WatermarkEvent.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/wartermark/WatermarkEvent.java
@@ -19,7 +19,7 @@
import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
-import io.debezium.util.SchemaNameAdjuster;
+import io.debezium.schema.SchemaNameAdjuster;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java
index 3136bb97d25..c8d4900b385 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java
@@ -22,9 +22,11 @@
import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.TemporaryClassLoaderContext;
import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.debezium.config.CommonConnectorConfig;
import io.debezium.connector.base.ChangeEventQueue;
import io.debezium.pipeline.DataChangeEvent;
import org.apache.kafka.connect.data.Struct;
@@ -89,7 +91,15 @@ public IncrementalSourceScanFetcher(FetchTask.Context taskContext, int subtaskId
public void submitTask(FetchTask fetchTask) {
this.snapshotSplitReadTask = fetchTask;
this.currentSnapshotSplit = fetchTask.getSplit().asSnapshotSplit();
- taskContext.configure(currentSnapshotSplit);
+ // Debezium 2.0 changed io.debezium.config.Instantiator to resolve classes through the
+ // thread context class loader. On a Flink task thread that loader can be a user code class
+ // loader left over from a previous job attempt, which is already closed, so Debezium fails
+ // with "Trying to access closed classloader" while building its own objects (topic naming
+ // strategy, transaction metadata factory, ...). Pin it to the loader that loaded Debezium.
+ try (TemporaryClassLoaderContext ignored =
+ TemporaryClassLoaderContext.of(CommonConnectorConfig.class.getClassLoader())) {
+ taskContext.configure(currentSnapshotSplit);
+ }
this.queue = taskContext.getQueue();
this.hasNextElement.set(true);
this.reachEnd.set(false);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java
index 937fefe67a1..d4d7c39bd12 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java
@@ -24,9 +24,11 @@
import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
import org.apache.flink.cdc.connectors.base.utils.SplitKeyUtils;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.TemporaryClassLoaderContext;
import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.debezium.config.CommonConnectorConfig;
import io.debezium.connector.base.ChangeEventQueue;
import io.debezium.pipeline.DataChangeEvent;
import io.debezium.relational.TableId;
@@ -89,7 +91,15 @@ public void submitTask(FetchTask fetchTask) {
this.streamFetchTask = fetchTask;
this.currentStreamSplit = fetchTask.getSplit().asStreamSplit();
configureFilter();
- taskContext.configure(currentStreamSplit);
+ // Debezium 2.0 changed io.debezium.config.Instantiator to resolve classes through the
+ // thread context class loader. On a Flink task thread that loader can be a user code class
+ // loader left over from a previous job attempt, which is already closed, so Debezium fails
+ // with "Trying to access closed classloader" while building its own objects (topic naming
+ // strategy, transaction metadata factory, ...). Pin it to the loader that loaded Debezium.
+ try (TemporaryClassLoaderContext ignored =
+ TemporaryClassLoaderContext.of(CommonConnectorConfig.class.getClassLoader())) {
+ taskContext.configure(currentStreamSplit);
+ }
this.queue = taskContext.getQueue();
startReadTask();
executorService.submit(
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/JdbcSourceFetchTaskContext.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/JdbcSourceFetchTaskContext.java
index 9975331e469..c35e7851183 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/JdbcSourceFetchTaskContext.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/JdbcSourceFetchTaskContext.java
@@ -35,7 +35,7 @@
import io.debezium.relational.RelationalDatabaseSchema;
import io.debezium.relational.Table;
import io.debezium.relational.TableId;
-import io.debezium.util.SchemaNameAdjuster;
+import io.debezium.schema.SchemaNameAdjuster;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SplitKeyUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SplitKeyUtils.java
index 94be44aa653..7d73b164289 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SplitKeyUtils.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SplitKeyUtils.java
@@ -20,7 +20,7 @@
import org.apache.flink.cdc.connectors.base.source.meta.split.FinishedSnapshotSplitInfo;
import org.apache.flink.table.types.logical.RowType;
-import io.debezium.util.SchemaNameAdjuster;
+import io.debezium.schema.SchemaNameAdjuster;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java
index 63da894ec85..ac4b73100a3 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java
@@ -7,19 +7,26 @@
package io.debezium.connector.db2;
import com.ibm.db2.jcc.DB2Driver;
+import io.debezium.DebeziumException;
+import io.debezium.config.CommonConnectorConfig;
import io.debezium.config.Configuration;
+import io.debezium.connector.db2.platform.Db2PlatformAdapter;
import io.debezium.jdbc.JdbcConfiguration;
import io.debezium.jdbc.JdbcConnection;
+import io.debezium.pipeline.spi.OffsetContext;
+import io.debezium.pipeline.spi.Partition;
import io.debezium.relational.Column;
import io.debezium.relational.ColumnEditor;
import io.debezium.relational.Table;
import io.debezium.relational.TableId;
import io.debezium.util.BoundedConcurrentHashMap;
import io.debezium.util.Collect;
+import org.apache.kafka.connect.errors.ConnectException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
@@ -28,14 +35,20 @@
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+/** {@link JdbcConnection} extension to be used with IBM Db2 */
/**
- * Copied from Debezium 1.9.8.Final. {@link JdbcConnection} extension to be used with IBM Db2
+ * Copied from Debezium project(2.7.4.Final).
*
- * @author Horia Chiorean (hchiorea@redhat.com), Jiri Pechanec, Peter Urbanetz
+ *
Change 1: override {@code resolveCatalogName} to return the real database name.
+ *
+ *
Change 2: override {@code readPrimaryKeyNames} / {@code readTableUniqueIndices} / {@code
+ * readTableNames} to query the JDBC metadata with the schema rather than the catalog, which is how
+ * Db2 exposes them.
*/
public class Db2Connection extends JdbcConnection {
@@ -44,48 +57,13 @@ public class Db2Connection extends JdbcConnection {
private static Logger LOGGER = LoggerFactory.getLogger(Db2Connection.class);
- private static final String CDC_SCHEMA = "ASNCDC";
-
private static final String STATEMENTS_PLACEHOLDER = "#";
- private static final String GET_MAX_LSN =
- "SELECT max(t.SYNCHPOINT) FROM ( SELECT CD_NEW_SYNCHPOINT AS SYNCHPOINT FROM "
- + CDC_SCHEMA
- + ".IBMSNAP_REGISTER UNION ALL SELECT SYNCHPOINT AS SYNCHPOINT FROM "
- + CDC_SCHEMA
- + ".IBMSNAP_REGISTER) t";
private static final String LOCK_TABLE = "SELECT * FROM # WITH CS"; // DB2
private static final String LSN_TO_TIMESTAMP =
"SELECT CURRENT TIMEstamp FROM sysibm.sysdummy1 WHERE ? > X'00000000000000000000000000000000'";
- private static final String GET_ALL_CHANGES_FOR_TABLE =
- "SELECT "
- + "CASE "
- + "WHEN IBMSNAP_OPERATION = 'D' AND (LEAD(cdc.IBMSNAP_OPERATION,1,'X') OVER (PARTITION BY cdc.IBMSNAP_COMMITSEQ ORDER BY cdc.IBMSNAP_INTENTSEQ)) ='I' THEN 3 "
- + "WHEN IBMSNAP_OPERATION = 'I' AND (LAG(cdc.IBMSNAP_OPERATION,1,'X') OVER (PARTITION BY cdc.IBMSNAP_COMMITSEQ ORDER BY cdc.IBMSNAP_INTENTSEQ)) ='D' THEN 4 "
- + "WHEN IBMSNAP_OPERATION = 'D' THEN 1 "
- + "WHEN IBMSNAP_OPERATION = 'I' THEN 2 "
- + "END "
- + "OPCODE,"
- + "cdc.* "
- + "FROM ASNCDC.# cdc WHERE IBMSNAP_COMMITSEQ >= ? AND IBMSNAP_COMMITSEQ <= ? "
- + "order by IBMSNAP_COMMITSEQ, IBMSNAP_INTENTSEQ";
-
- private static final String GET_LIST_OF_CDC_ENABLED_TABLES =
- "select r.SOURCE_OWNER, r.SOURCE_TABLE, r.CD_OWNER, r.CD_TABLE, r.CD_NEW_SYNCHPOINT, r.CD_OLD_SYNCHPOINT, t.TBSPACEID, t.TABLEID , CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER )from "
- + CDC_SCHEMA
- + ".IBMSNAP_REGISTER r left JOIN SYSCAT.TABLES t ON r.SOURCE_OWNER = t.TABSCHEMA AND r.SOURCE_TABLE = t.TABNAME WHERE r.SOURCE_OWNER <> ''";
-
- // No new Tabels 1=0
- private static final String GET_LIST_OF_NEW_CDC_ENABLED_TABLES =
- "select CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER ) AS OBJECTID, "
- + " CD_OWNER CONCAT '.' CONCAT CD_TABLE, "
- + " CD_NEW_SYNCHPOINT, "
- + " CD_OLD_SYNCHPOINT "
- + "from ASNCDC.IBMSNAP_REGISTER r left JOIN SYSCAT.TABLES t ON r.SOURCE_OWNER = t.TABSCHEMA AND r.SOURCE_TABLE = t.TABNAME "
- + "WHERE r.SOURCE_OWNER <> '' AND 1=0 AND CD_NEW_SYNCHPOINT > ? AND CD_OLD_SYNCHPOINT < ? ";
-
private static final String GET_LIST_OF_KEY_COLUMNS =
"SELECT "
+ "CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER ) as objectid, "
@@ -123,15 +101,21 @@ public class Db2Connection extends JdbcConnection {
private final BoundedConcurrentHashMap lsnToInstantCache;
+ private final Db2ConnectorConfig connectorConfig;
+ private final Db2PlatformAdapter platform;
+
/**
* Creates a new connection using the supplied configuration.
*
* @param config {@link Configuration} instance, may not be null.
*/
- public Db2Connection(JdbcConfiguration config) {
- super(config, FACTORY, QUOTED_CHARACTER, QUOTED_CHARACTER);
+ public Db2Connection(Db2ConnectorConfig config) {
+ super(config.getJdbcConfig(), FACTORY, QUOTED_CHARACTER, QUOTED_CHARACTER);
+
+ connectorConfig = config;
lsnToInstantCache = new BoundedConcurrentHashMap<>(100);
realDatabaseName = retrieveRealDatabaseName();
+ platform = connectorConfig.getDb2Platform().createAdapter(connectorConfig);
}
/**
@@ -139,7 +123,7 @@ public Db2Connection(JdbcConfiguration config) {
*/
public Lsn getMaxLsn() throws SQLException {
return queryAndMap(
- GET_MAX_LSN,
+ platform.getMaxLsnQuery(),
singleResultMapper(
rs -> {
final Lsn ret = Lsn.valueOf(rs.getBytes(1));
@@ -162,7 +146,8 @@ public void getChangesForTable(
TableId tableId, Lsn fromLsn, Lsn toLsn, ResultSetConsumer consumer)
throws SQLException {
final String query =
- GET_ALL_CHANGES_FOR_TABLE.replace(STATEMENTS_PLACEHOLDER, cdcNameForTable(tableId));
+ platform.getAllChangesForTableQuery()
+ .replace(STATEMENTS_PLACEHOLDER, cdcNameForTable(tableId));
prepareQuery(
query,
statement -> {
@@ -193,8 +178,8 @@ public void getChangesForTables(
int idx = 0;
for (Db2ChangeTable changeTable : changeTables) {
final String query =
- GET_ALL_CHANGES_FOR_TABLE.replace(
- STATEMENTS_PLACEHOLDER, changeTable.getCaptureInstance());
+ platform.getAllChangesForTableQuery()
+ .replace(STATEMENTS_PLACEHOLDER, changeTable.getCaptureInstance());
queries[idx] = query;
// If the table was added in the middle of queried buffer we need
// to adjust from to the first LSN available
@@ -263,10 +248,10 @@ public Instant timestampOfLsn(Lsn lsn) throws SQLException {
}
@Override
- public Optional getCurrentTimestamp() throws SQLException {
+ public Optional getCurrentTimestamp() throws SQLException {
return queryAndMap(
"SELECT CURRENT_TIMESTAMP result FROM sysibm.sysdummy1",
- rs -> rs.next() ? Optional.of(rs.getTimestamp(1)) : Optional.empty());
+ rs -> rs.next() ? Optional.of(rs.getTimestamp(1).toInstant()) : Optional.empty());
}
/**
@@ -309,10 +294,9 @@ public Lsn getFromLsn() {
}
public Set listOfChangeTables() throws SQLException {
- final String query = GET_LIST_OF_CDC_ENABLED_TABLES;
return queryAndMap(
- query,
+ platform.getListOfCdcEnabledTablesQuery(),
rs -> {
final Set changeTables = new HashSet<>();
while (rs.next()) {
@@ -325,22 +309,21 @@ public Set listOfChangeTables() throws SQLException {
*/
changeTables.add(
new Db2ChangeTable(
- new TableId(
- realDatabaseName, rs.getString(1), rs.getString(2)),
+ new TableId("", rs.getString(1), rs.getString(2)),
rs.getString(4),
rs.getInt(9),
Lsn.valueOf(rs.getBytes(5)),
- Lsn.valueOf(rs.getBytes(6))));
+ Lsn.valueOf(rs.getBytes(6)),
+ connectorConfig.getCdcChangeTablesSchema()));
}
return changeTables;
});
}
public Set listOfNewChangeTables(Lsn fromLsn, Lsn toLsn) throws SQLException {
- final String query = GET_LIST_OF_NEW_CDC_ENABLED_TABLES;
return prepareQueryAndMap(
- query,
+ platform.getListOfNewCdcEnabledTablesQuery(),
ps -> {
ps.setBytes(1, fromLsn.getBinary());
ps.setBytes(2, toLsn.getBinary());
@@ -353,7 +336,8 @@ public Set listOfNewChangeTables(Lsn fromLsn, Lsn toLsn) throws
rs.getString(2),
rs.getInt(1),
Lsn.valueOf(rs.getBytes(3)),
- Lsn.valueOf(rs.getBytes(4))));
+ Lsn.valueOf(rs.getBytes(4)),
+ connectorConfig.getCdcChangeTablesSchema()));
}
return changeTables;
});
@@ -469,6 +453,13 @@ public String connectionString() {
return connectionString(URL_PATTERN);
}
+ @Override
+ public Optional nullsSortLast() {
+ // "The null value is higher than all other values"
+ // https://www.ibm.com/docs/en/db2/11.5?topic=subselect-order-by-clause
+ return Optional.of(true);
+ }
+
@Override
public String quotedTableIdString(TableId tableId) {
StringBuilder quoted = new StringBuilder();
@@ -479,6 +470,196 @@ public String quotedTableIdString(TableId tableId) {
return quoted.toString();
}
+ @Override
+ public JdbcConnection prepareQuery(
+ String[] multiQuery,
+ StatementPreparer[] preparers,
+ BlockingMultiResultSetConsumer resultConsumer)
+ throws SQLException, InterruptedException {
+ final ResultSet[] resultSets = new ResultSet[multiQuery.length];
+ final PreparedStatement[] preparedStatements = new PreparedStatement[multiQuery.length];
+
+ try {
+ for (int i = 0; i < multiQuery.length; i++) {
+ final String query = multiQuery[i];
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("running '{}'", query);
+ }
+ // Purposely create the statement this way
+ final PreparedStatement statement = createPreparedStatement(query);
+ preparedStatements[i] = statement;
+ preparers[i].accept(statement);
+ resultSets[i] = statement.executeQuery();
+ }
+ if (resultConsumer != null) {
+ resultConsumer.accept(resultSets);
+ }
+ } finally {
+ for (ResultSet rs : resultSets) {
+ if (rs != null) {
+ try {
+ rs.close();
+ } catch (Exception ei) {
+ }
+ }
+ }
+ // Db2 requires closing prepared statements to avoid caching result-set column
+ // structures
+ for (PreparedStatement ps : preparedStatements) {
+ closePreparedStatement(ps);
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public JdbcConnection prepareQueryWithBlockingConsumer(
+ String preparedQueryString,
+ StatementPreparer preparer,
+ BlockingResultSetConsumer resultConsumer)
+ throws SQLException, InterruptedException {
+ // Db2 requires closing prepared statements to avoid caching result-set column structures
+ try (PreparedStatement statement = createPreparedStatement(preparedQueryString)) {
+ preparer.accept(statement);
+ try (ResultSet resultSet = statement.executeQuery(); ) {
+ if (resultConsumer != null) {
+ resultConsumer.accept(resultSet);
+ }
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public JdbcConnection prepareQuery(String preparedQueryString) throws SQLException {
+ // Db2 requires closing prepared statements to avoid caching result-set column structures
+ try (PreparedStatement statement = createPreparedStatement(preparedQueryString)) {
+ statement.executeQuery();
+ }
+ return this;
+ }
+
+ @Override
+ public JdbcConnection prepareQuery(
+ String preparedQueryString,
+ StatementPreparer preparer,
+ ResultSetConsumer resultConsumer)
+ throws SQLException {
+ // Db2 requires closing prepared statements to avoid caching result-set column structures
+ try (PreparedStatement statement = createPreparedStatement(preparedQueryString)) {
+ preparer.accept(statement);
+ try (ResultSet resultSet = statement.executeQuery(); ) {
+ if (resultConsumer != null) {
+ resultConsumer.accept(resultSet);
+ }
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public T prepareQueryAndMap(
+ String preparedQueryString, StatementPreparer preparer, ResultSetMapper mapper)
+ throws SQLException {
+ Objects.requireNonNull(mapper, "Mapper must be provided");
+ // Db2 requires closing prepared statements to avoid caching result-set column structures
+ try (PreparedStatement statement = createPreparedStatement(preparedQueryString)) {
+ preparer.accept(statement);
+ try (ResultSet resultSet = statement.executeQuery(); ) {
+ return mapper.apply(resultSet);
+ }
+ }
+ }
+
+ @Override
+ public JdbcConnection prepareUpdate(String stmt, StatementPreparer preparer)
+ throws SQLException {
+ // Db2 requires closing prepared statements to avoid caching result-set column structures
+ try (PreparedStatement statement = createPreparedStatement(stmt)) {
+ if (preparer != null) {
+ preparer.accept(statement);
+ }
+ LOGGER.trace("Executing statement '{}'", stmt);
+ statement.execute();
+ }
+ return this;
+ }
+
+ @Override
+ public JdbcConnection prepareQuery(
+ String preparedQueryString,
+ List> parameters,
+ ParameterResultSetConsumer resultConsumer)
+ throws SQLException {
+ // Db2 requires closing prepared statements to avoid caching result-set column structures
+ try (PreparedStatement statement = createPreparedStatement(preparedQueryString)) {
+ int index = 1;
+ for (final Object parameter : parameters) {
+ statement.setObject(index++, parameter);
+ }
+ try (ResultSet resultSet = statement.executeQuery()) {
+ if (resultConsumer != null) {
+ resultConsumer.accept(parameters, resultSet);
+ }
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public TableId createTableId(String databaseName, String schemaName, String tableName) {
+ return new TableId(null, schemaName, tableName);
+ }
+
+ public boolean validateLogPosition(
+ Partition partition, OffsetContext offset, CommonConnectorConfig config) {
+
+ final Lsn storedLsn = ((Db2OffsetContext) offset).getChangePosition().getCommitLsn();
+
+ String oldestFirstChangeQuery =
+ String.format(
+ "SELECT min(RESTART_SEQ) FROM %s.IBMSNAP_CAPMON;",
+ connectorConfig.getCdcControlSchema());
+
+ try {
+ final String oldestScn =
+ singleOptionalValue(oldestFirstChangeQuery, rs -> rs.getString(1));
+
+ if (oldestScn == null) {
+ return false;
+ }
+
+ LOGGER.trace("Oldest SCN in logs is '{}'", oldestScn);
+ return storedLsn == null || Lsn.valueOf(oldestScn).compareTo(storedLsn) < 0;
+ } catch (SQLException e) {
+ throw new DebeziumException("Unable to get last available log position", e);
+ }
+ }
+
+ public T singleOptionalValue(String query, ResultSetExtractor extractor)
+ throws SQLException {
+ return queryAndMap(query, rs -> rs.next() ? extractor.apply(rs) : null);
+ }
+
+ private PreparedStatement createPreparedStatement(String query) {
+ try {
+ LOGGER.trace("Creating prepared statement '{}'", query);
+ return connection().prepareStatement(query);
+ } catch (SQLException e) {
+ throw new ConnectException(e);
+ }
+ }
+
+ private void closePreparedStatement(PreparedStatement statement) {
+ if (statement != null) {
+ try {
+ statement.close();
+ } catch (SQLException e) {
+ // ignored
+ }
+ }
+ }
+
protected String resolveCatalogName(String catalogName) {
return realDatabaseName;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java
index 261a59fb2f2..fd75715d706 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java
@@ -10,9 +10,11 @@
import io.debezium.pipeline.EventDispatcher;
import io.debezium.pipeline.source.spi.ChangeTableResultSet;
import io.debezium.pipeline.source.spi.StreamingChangeEventSource;
+import io.debezium.relational.Table;
import io.debezium.relational.TableId;
import io.debezium.schema.DatabaseSchema;
import io.debezium.schema.SchemaChangeEvent.SchemaChangeEventType;
+import io.debezium.snapshot.SnapshotterService;
import io.debezium.util.Clock;
import io.debezium.util.Metronome;
import org.slf4j.Logger;
@@ -35,27 +37,24 @@
import java.util.stream.Collectors;
/**
- * Copied from Debezium project(1.9.8.final)
+ * Copied from Debezium project(2.7.4.Final)..
*
- *
A {@link StreamingChangeEventSource} based on DB2 change data capture functionality. A main
- * loop polls database DDL change and change data tables and turns them into change events.
+ *
Change 1: the effective offset context is initialised at the top of {@code execute} —
+ * Debezium's coordinator calls {@code init(offsetContext)} before {@code execute}, but Flink CDC
+ * drives this source directly and never calls {@code init}.
*
- *
The connector uses CDC functionality of DB2 that is implemented as as a process that monitors
- * source table and write changes from the table into the change table.
+ *
Change 2: add the {@code afterHandleLsn} hook, called after each iteration commits its
+ * position, so Flink CDC's bounded stream task can stop at the high watermark.
*
- *
The main loop keeps a pointer to the LSN of changes that were already processed. It queries
- * all change tables and get result set of changes. It always finds the smallest LSN across all
- * tables and the change is converted into the event message and sent downstream. The process
- * repeats until all result sets are empty. The LSN is marked and the procedure repeats.
- *
- *
The schema changes detection follows the procedure recommended by DB2 CDC documentation. The
- * database operator should create one more capture process (and table) when a table schema is
- * updated. The code detects presence of two change tables for a single source table. It decides
- * which table is the new one depending on LSNs stored in them. The loop streams changes from the
- * older table till there are events in new table with the LSN larger than in the old one. Then the
- * change table is switched and streaming is executed from the new one.
- *
- * @author Jiri Pechanec, Peter Urbanetz
+ *
Change 3: keep the pre-2.7 exclusive upper bound when deciding which capture instances
+ * represent a schema change. Upstream 2.7 widened this to {@code currentMaxLsn.increment()}, making
+ * the window inclusive. Flink CDC reads a change table's {@code startLsn} from a column that tracks
+ * the capture progress rather than a fixed creation LSN, so {@code startLsn} equals {@code
+ * currentMaxLsn} on virtually every poll. With the inclusive bound the same capture instance is
+ * queued as a schema change over and over, and {@code migrateTable} re-reads the live source table
+ * schema each time. After an {@code ALTER TABLE ... ADD COLUMN} that re-read installs the new,
+ * wider schema while the still-open change-table result set only yields the old column count,
+ * producing "Data row is smaller than a column index".
*/
public class Db2StreamingChangeEventSource
implements StreamingChangeEventSource {
@@ -86,6 +85,9 @@ public class Db2StreamingChangeEventSource
private final Db2DatabaseSchema schema;
private final Duration pollInterval;
private final Db2ConnectorConfig connectorConfig;
+ private Db2OffsetContext effectiveOffsetContext;
+
+ private final SnapshotterService snapshotterService;
public Db2StreamingChangeEventSource(
Db2ConnectorConfig connectorConfig,
@@ -94,7 +96,8 @@ public Db2StreamingChangeEventSource(
EventDispatcher dispatcher,
ErrorHandler errorHandler,
Clock clock,
- Db2DatabaseSchema schema) {
+ Db2DatabaseSchema schema,
+ SnapshotterService snapshotterService) {
this.connectorConfig = connectorConfig;
this.dataConnection = dataConnection;
this.metadataConnection = metadataConnection;
@@ -103,6 +106,15 @@ public Db2StreamingChangeEventSource(
this.clock = clock;
this.schema = schema;
this.pollInterval = connectorConfig.getPollInterval();
+ this.snapshotterService = snapshotterService;
+ }
+
+ public void init(Db2OffsetContext offsetContext) {
+
+ this.effectiveOffsetContext =
+ offsetContext != null
+ ? offsetContext
+ : new Db2OffsetContext(connectorConfig, TxLogPosition.NULL, false, false);
}
@Override
@@ -111,9 +123,11 @@ public void execute(
Db2Partition partition,
Db2OffsetContext offsetContext)
throws InterruptedException {
- if (!connectorConfig.getSnapshotMode().shouldStream()) {
- LOGGER.info("Streaming is not enabled in current configuration");
- return;
+ // Debezium's ChangeEventSourceCoordinator calls init(offsetContext) before execute(...) to
+ // set the effective offset context. Flink CDC drives this source directly and never calls
+ // init, so fall back to the offset context it passes in.
+ if (getOffsetContext() == null) {
+ init(offsetContext);
}
final Metronome metronome = Metronome.sleeper(pollInterval, clock);
@@ -142,7 +156,7 @@ public void execute(
// situation
if (!currentMaxLsn.isAvailable()) {
LOGGER.warn(
- "No maximum LSN recorded in the database; please ensure that the DB2 Agent is running");
+ "No maximum LSN in the database; please ensure that the DB2 Agent is running");
metronome.pause();
continue;
}
@@ -338,7 +352,8 @@ public void execute(
operation,
data,
dataNext,
- clock));
+ clock,
+ connectorConfig));
tableWithSmallestLsn.next();
}
});
@@ -350,12 +365,24 @@ public void execute(
} catch (SQLException e) {
tablesSlot.set(processErrorFromChangeTableQuery(e, tablesSlot.get()));
}
+
+ if (context.isPaused()) {
+ LOGGER.info("Streaming will now pause");
+ context.streamingPaused();
+ context.waitSnapshotCompletion();
+ LOGGER.info("Streaming resumed");
+ }
}
} catch (Exception e) {
errorHandler.setProducerThrowable(e);
}
}
+ @Override
+ public Db2OffsetContext getOffsetContext() {
+ return effectiveOffsetContext;
+ }
+
private void migrateTable(
Db2Partition partition,
Db2OffsetContext offsetContext,
@@ -363,15 +390,21 @@ private void migrateTable(
throws InterruptedException, SQLException {
final Db2ChangeTable newTable = schemaChangeCheckpoints.poll();
LOGGER.info("Migrating schema to {}", newTable);
+ Table tableSchema = metadataConnection.getTableSchemaFromTable(newTable);
+ offsetContext.event(newTable.getSourceTableId(), Instant.now());
dispatcher.dispatchSchemaChangeEvent(
partition,
+ offsetContext,
newTable.getSourceTableId(),
new Db2SchemaChangeEventEmitter(
partition,
offsetContext,
newTable,
- metadataConnection.getTableSchemaFromTable(newTable),
+ tableSchema,
+ schema,
SchemaChangeEventType.ALTER));
+
+ newTable.setSourceTable(tableSchema);
}
private Db2ChangeTable[] processErrorFromChangeTableQuery(
@@ -450,12 +483,14 @@ private Db2ChangeTable[] getCdcTablesToQuery(
// obtained from change table
dispatcher.dispatchSchemaChangeEvent(
partition,
+ offsetContext,
currentTable.getSourceTableId(),
new Db2SchemaChangeEventEmitter(
partition,
offsetContext,
currentTable,
dataConnection.getTableSchemaFromTable(currentTable),
+ schema,
SchemaChangeEventType.CREATE));
}
tables.add(currentTable);
@@ -470,13 +505,11 @@ private Db2ChangeTable[] getCdcTablesToQuery(
* changes across all tables.
* This class represents an open database cursor over the change table that is able to move the
* cursor forward and report the LSN for the change to which the cursor now points.
- *
- * @author Jiri Pechanec
*/
private static class ChangeTablePointer
extends ChangeTableResultSet {
- public ChangeTablePointer(Db2ChangeTable changeTable, ResultSet resultSet) {
+ ChangeTablePointer(Db2ChangeTable changeTable, ResultSet resultSet) {
super(changeTable, resultSet, COL_DATA);
}
@@ -495,7 +528,7 @@ protected TxLogPosition getNextChangePosition(ResultSet resultSet) throws SQLExc
}
}
- /** expose control to the user to stop the connector. */
+ /** Expose control to the subclass to stop the connector. */
protected void afterHandleLsn(Db2Partition partition, Lsn toLsn) {
// do nothing
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java
index 2b704562d45..1a2b0c9e174 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java
@@ -61,8 +61,9 @@ public DebeziumSourceFunction build() {
props.setProperty("database.user", checkNotNull(username));
props.setProperty("database.password", checkNotNull(password));
props.setProperty("database.dbname", checkNotNull(database));
- props.setProperty("database.server.name", DB2_DATABASE_SERVER_NAME); // Hard-coded here
- props.setProperty("database.history.skip.unparseable.ddl", String.valueOf(true));
+ // Debezium 2.0 renamed "database.server.name" to "topic.prefix".
+ props.setProperty("topic.prefix", DB2_DATABASE_SERVER_NAME); // Hard-coded here
+ props.setProperty("schema.history.internal.skip.unparseable.ddl", String.valueOf(true));
if (tableList != null) {
props.setProperty("table.include.list", String.join(",", tableList));
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
index ab35aa409ea..ee3b1018545 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
@@ -19,6 +19,7 @@
import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.util.TemporaryClassLoaderContext;
import io.debezium.config.Configuration;
import io.debezium.connector.db2.Db2ConnectorConfig;
@@ -89,6 +90,13 @@ public Db2SourceConfig(
@Override
public Db2ConnectorConfig getDbzConnectorConfig() {
- return new Db2ConnectorConfig(getDbzConfiguration());
+ // Debezium 2.0 resolves classes through the thread context class loader
+ // (io.debezium.config.Instantiator, and the ServiceLoader based SPIs). On a Flink
+ // thread that loader may be a user code class loader from a previous job attempt,
+ // which is already closed. Pin it to the loader that loaded Debezium.
+ try (TemporaryClassLoaderContext ignored =
+ TemporaryClassLoaderContext.of(Db2ConnectorConfig.class.getClassLoader())) {
+ return new Db2ConnectorConfig(getDbzConfiguration());
+ }
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
index 365e7871099..82365e84075 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
@@ -43,8 +43,9 @@ public Db2SourceConfig create(int subtask) {
// set database history impl to flink database history
props.setProperty(
- "database.history", EmbeddedFlinkDatabaseHistory.class.getCanonicalName());
- props.setProperty("database.history.instance.name", UUID.randomUUID() + "_" + subtask);
+ "schema.history.internal", EmbeddedFlinkDatabaseHistory.class.getCanonicalName());
+ props.setProperty(
+ "schema.history.internal.instance.name", UUID.randomUUID() + "_" + subtask);
// hard code server name, because we don't need to distinguish it, docs:
// Logical name that identifies and provides a namespace for the SQL Server database
@@ -52,12 +53,13 @@ public Db2SourceConfig create(int subtask) {
// all other connectors, since it is used as a prefix for all Kafka topic names
// emanating from this connector. Only alphanumeric characters and underscores should be
// used.
- props.setProperty("database.server.name", DATABASE_SERVER_NAME);
+ // Debezium 2.0 renamed "database.server.name" to "topic.prefix".
+ props.setProperty("topic.prefix", DATABASE_SERVER_NAME);
props.setProperty("database.hostname", checkNotNull(hostname));
props.setProperty("database.user", checkNotNull(username));
props.setProperty("database.password", checkNotNull(password));
props.setProperty("database.port", String.valueOf(port));
- props.setProperty("database.history.skip.unparseable.ddl", String.valueOf(true));
+ props.setProperty("schema.history.internal.skip.unparseable.ddl", String.valueOf(true));
props.setProperty("database.dbname", checkNotNull(databaseList.get(0)));
if (tableList != null) {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java
index d6baf843a5c..49233040779 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java
@@ -21,17 +21,22 @@
import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
import org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask;
import org.apache.flink.cdc.connectors.db2.source.fetch.Db2StreamFetchTask.StreamSplitReadTask;
+import org.apache.flink.cdc.debezium.internal.SnapshotterServiceFactory;
import io.debezium.DebeziumException;
import io.debezium.config.Configuration;
import io.debezium.connector.db2.Db2Connection;
+import io.debezium.connector.db2.Db2Connector;
import io.debezium.connector.db2.Db2ConnectorConfig;
import io.debezium.connector.db2.Db2DatabaseSchema;
import io.debezium.connector.db2.Db2OffsetContext;
import io.debezium.connector.db2.Db2Partition;
import io.debezium.heartbeat.Heartbeat;
import io.debezium.pipeline.EventDispatcher;
+import io.debezium.pipeline.notification.NotificationService;
+import io.debezium.pipeline.signal.actions.snapshotting.SnapshotConfiguration;
import io.debezium.pipeline.source.AbstractSnapshotChangeEventSource;
+import io.debezium.pipeline.source.SnapshottingTask;
import io.debezium.pipeline.source.spi.ChangeEventSource;
import io.debezium.pipeline.source.spi.SnapshotProgressListener;
import io.debezium.pipeline.spi.ChangeRecordEmitter;
@@ -40,6 +45,7 @@
import io.debezium.relational.SnapshotChangeRecordEmitter;
import io.debezium.relational.Table;
import io.debezium.relational.TableId;
+import io.debezium.schema.SchemaFactory;
import io.debezium.util.Clock;
import io.debezium.util.ColumnUtils;
import io.debezium.util.Strings;
@@ -52,6 +58,7 @@
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
+import java.util.Collections;
import static org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils.buildSplitScanQuery;
import static org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils.readTableSplitDataStatement;
@@ -82,7 +89,10 @@ protected void executeDataSnapshot(Context context) throws Exception {
snapshotSplitReadTask.execute(
changeEventSourceContext,
sourceFetchContext.getPartition(),
- sourceFetchContext.getOffsetContext());
+ sourceFetchContext.getOffsetContext(),
+ snapshotSplitReadTask.getSnapshottingTask(
+ sourceFetchContext.getPartition(),
+ sourceFetchContext.getOffsetContext()));
// execute stream read task
if (!snapshotResult.isCompletedOrSkipped()) {
taskRunning = false;
@@ -127,14 +137,16 @@ private StreamSplitReadTask createBackFillLsnSplitReadTask(
.with(Heartbeat.HEARTBEAT_INTERVAL, 0)
.build();
// task to read wal and backfill for current split
+ Db2ConnectorConfig backfillConnectorConfig = new Db2ConnectorConfig(dezConf);
return new StreamSplitReadTask(
- new Db2ConnectorConfig(dezConf),
+ backfillConnectorConfig,
context.getConnection(),
context.getMetaDataConnection(),
context.getEventDispatcher(),
context.getWaterMarkDispatcher(),
context.getErrorHandler(),
context.getDatabaseSchema(),
+ SnapshotterServiceFactory.create(backfillConnectorConfig, Db2Connector.class),
backfillBinlogSplit);
}
@@ -166,7 +178,14 @@ public Db2SnapshotSplitReadTask(
EventDispatcher dispatcher,
EventDispatcher.SnapshotReceiver snapshotReceiver,
SnapshotSplit snapshotSplit) {
- super(connectorConfig, snapshotProgressListener);
+ super(
+ connectorConfig,
+ snapshotProgressListener,
+ new NotificationService<>(
+ Collections.emptyList(),
+ connectorConfig,
+ SchemaFactory.get(),
+ notification -> {}));
this.offsetContext = previousOffset;
this.connectorConfig = connectorConfig;
this.databaseSchema = databaseSchema;
@@ -182,12 +201,12 @@ public Db2SnapshotSplitReadTask(
public SnapshotResult execute(
ChangeEventSourceContext context,
Db2Partition partition,
- Db2OffsetContext previousOffset)
+ Db2OffsetContext previousOffset,
+ SnapshottingTask snapshottingTask)
throws InterruptedException {
- SnapshottingTask snapshottingTask = getSnapshottingTask(partition, previousOffset);
final Db2SnapshotContext ctx;
try {
- ctx = prepare(partition);
+ ctx = prepare(partition, false);
} catch (Exception e) {
LOG.error("Failed to initialize snapshot context.", e);
throw new RuntimeException(e);
@@ -218,14 +237,26 @@ protected SnapshotResult doExecute(
}
@Override
- protected SnapshottingTask getSnapshottingTask(
+ public SnapshottingTask getSnapshottingTask(
Db2Partition partition, Db2OffsetContext previousOffset) {
- return new SnapshottingTask(false, true);
+ return new SnapshottingTask(
+ false, true, Collections.emptyList(), Collections.emptyMap(), false);
}
@Override
- protected Db2SnapshotContext prepare(Db2Partition partition) throws Exception {
- return new Db2SnapshotContext(partition);
+ public SnapshottingTask getBlockingSnapshottingTask(
+ Db2Partition partition,
+ Db2OffsetContext previousOffset,
+ SnapshotConfiguration snapshotConfiguration) {
+ // Debezium 2.6 made this abstract. Flink CDC drives its own split snapshot and never
+ // runs Debezium's signal-based blocking snapshot, so it behaves like the regular one.
+ return getSnapshottingTask(partition, previousOffset);
+ }
+
+ @Override
+ protected Db2SnapshotContext prepare(Db2Partition partition, boolean onDemand)
+ throws Exception {
+ return new Db2SnapshotContext(partition, onDemand);
}
private void createDataEvents(Db2SnapshotContext snapshotContext, TableId tableId)
@@ -279,8 +310,7 @@ private void createDataEventsForTable(
while (rs.next()) {
rows++;
- final Object[] row =
- jdbcConnection.rowToArray(table, databaseSchema, rs, columnArray);
+ final Object[] row = jdbcConnection.rowToArray(table, rs, columnArray);
if (logTimer.expired()) {
long stop = clock.currentTimeInMillis();
LOG.info(
@@ -312,7 +342,7 @@ protected ChangeRecordEmitter getChangeRecordEmitter(
Db2SnapshotContext snapshotContext, TableId tableId, Object[] row) {
snapshotContext.offset.event(tableId, clock.currentTime());
return new SnapshotChangeRecordEmitter<>(
- snapshotContext.partition, snapshotContext.offset, row, clock);
+ snapshotContext.partition, snapshotContext.offset, row, clock, connectorConfig);
}
private Threads.Timer getTableScanLogTimer() {
@@ -323,8 +353,9 @@ private static class Db2SnapshotContext
extends RelationalSnapshotChangeEventSource.RelationalSnapshotContext<
Db2Partition, Db2OffsetContext> {
- public Db2SnapshotContext(Db2Partition partition) throws SQLException {
- super(partition, "");
+ public Db2SnapshotContext(Db2Partition partition, boolean onDemand)
+ throws SQLException {
+ super(partition, "", onDemand);
}
}
}
@@ -344,5 +375,25 @@ public void finished() {
public boolean isRunning() {
return taskRunning;
}
+
+ // The following methods are only used by Debezium's signal-based blocking snapshot,
+ // which Flink CDC does not use, so they are no-ops here.
+
+ @Override
+ public boolean isPaused() {
+ return false;
+ }
+
+ @Override
+ public void resumeStreaming() throws InterruptedException {}
+
+ @Override
+ public void waitSnapshotCompletion() throws InterruptedException {}
+
+ @Override
+ public void streamingPaused() {}
+
+ @Override
+ public void waitStreamingPaused() throws InterruptedException {}
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java
index 64e99c32fd2..a4b3bf4122a 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java
@@ -31,6 +31,7 @@
import org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils;
import org.apache.flink.table.types.logical.RowType;
+import io.debezium.config.CommonConnectorConfig;
import io.debezium.connector.base.ChangeEventQueue;
import io.debezium.connector.base.ChangeEventQueue.Builder;
import io.debezium.connector.db2.Db2Connection;
@@ -41,7 +42,6 @@
import io.debezium.connector.db2.Db2OffsetContext.Loader;
import io.debezium.connector.db2.Db2Partition;
import io.debezium.connector.db2.Db2TaskContext;
-import io.debezium.connector.db2.Db2TopicSelector;
import io.debezium.connector.db2.SourceInfo;
import io.debezium.data.Envelope.FieldName;
import io.debezium.pipeline.DataChangeEvent;
@@ -56,10 +56,10 @@
import io.debezium.relational.Table;
import io.debezium.relational.TableId;
import io.debezium.relational.Tables.TableFilter;
-import io.debezium.schema.DataCollectionId;
-import io.debezium.schema.TopicSelector;
+import io.debezium.schema.SchemaNameAdjuster;
+import io.debezium.spi.schema.DataCollectionId;
+import io.debezium.spi.topic.TopicNamingStrategy;
import io.debezium.util.Collect;
-import io.debezium.util.SchemaNameAdjuster;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
@@ -82,7 +82,7 @@ public class Db2SourceFetchTaskContext extends JdbcSourceFetchTaskContext {
private ErrorHandler errorHandler;
private ChangeEventQueue queue;
private Db2TaskContext taskContext;
- private TopicSelector topicSelector;
+ private TopicNamingStrategy topicSelector;
private EventDispatcher.SnapshotReceiver snapshotReceiver;
private SnapshotChangeEventSourceMetrics snapshotChangeEventSourceMetrics;
private StreamingChangeEventSourceMetrics streamingChangeEventSourceMetrics;
@@ -102,7 +102,8 @@ public Db2SourceFetchTaskContext(
public void configure(SourceSplitBase sourceSplitBase) {
// initial stateful objects
final Db2ConnectorConfig connectorConfig = getDbzConnectorConfig();
- this.topicSelector = Db2TopicSelector.defaultSelector(connectorConfig);
+ this.topicSelector =
+ connectorConfig.getTopicNamingStrategy(CommonConnectorConfig.TOPIC_NAMING_STRATEGY);
EmbeddedFlinkDatabaseHistory.registerHistory(
sourceConfig
.getDbzConfiguration()
@@ -156,7 +157,7 @@ public void configure(SourceSplitBase sourceSplitBase) {
this.streamingChangeEventSourceMetrics =
changeEventSourceMetricsFactory.getStreamingMetrics(
taskContext, queue, metadataProvider);
- this.errorHandler = new ErrorHandler(Db2Connector.class, connectorConfig, queue);
+ this.errorHandler = new ErrorHandler(Db2Connector.class, connectorConfig, queue, null);
}
/** Loads the connector's persistent offset (if present) via the given loader. */
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java
index 43581c54917..aa9b7c6b6a9 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java
@@ -24,9 +24,11 @@
import org.apache.flink.cdc.connectors.base.source.meta.wartermark.WatermarkKind;
import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
import org.apache.flink.cdc.connectors.db2.source.offset.LsnOffset;
+import org.apache.flink.cdc.debezium.internal.SnapshotterServiceFactory;
import io.debezium.DebeziumException;
import io.debezium.connector.db2.Db2Connection;
+import io.debezium.connector.db2.Db2Connector;
import io.debezium.connector.db2.Db2ConnectorConfig;
import io.debezium.connector.db2.Db2DatabaseSchema;
import io.debezium.connector.db2.Db2OffsetContext;
@@ -37,6 +39,7 @@
import io.debezium.pipeline.EventDispatcher;
import io.debezium.pipeline.source.spi.ChangeEventSource.ChangeEventSourceContext;
import io.debezium.relational.TableId;
+import io.debezium.snapshot.SnapshotterService;
import io.debezium.util.Clock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,6 +70,8 @@ public void execute(Context context) throws Exception {
sourceFetchContext.getWaterMarkDispatcher(),
sourceFetchContext.getErrorHandler(),
sourceFetchContext.getDatabaseSchema(),
+ SnapshotterServiceFactory.create(
+ sourceFetchContext.getDbzConnectorConfig(), Db2Connector.class),
split);
RedoLogSplitChangeEventSourceContext changeEventSourceContext =
new RedoLogSplitChangeEventSourceContext();
@@ -111,6 +116,7 @@ public StreamSplitReadTask(
WatermarkDispatcher watermarkDispatcher,
ErrorHandler errorHandler,
Db2DatabaseSchema schema,
+ SnapshotterService snapshotterService,
StreamSplit lsnSplit) {
super(
connectorConfig,
@@ -119,7 +125,8 @@ public StreamSplitReadTask(
eventDispatcher,
errorHandler,
Clock.system(),
- schema);
+ schema,
+ snapshotterService);
this.lsnSplit = lsnSplit;
this.watermarkDispatcher = watermarkDispatcher;
this.errorHandler = errorHandler;
@@ -173,5 +180,25 @@ private class RedoLogSplitChangeEventSourceContext implements ChangeEventSourceC
public boolean isRunning() {
return taskRunning;
}
+
+ // The following methods are only used by Debezium's signal-based blocking snapshot,
+ // which Flink CDC does not use, so they are no-ops here.
+
+ @Override
+ public boolean isPaused() {
+ return false;
+ }
+
+ @Override
+ public void resumeStreaming() throws InterruptedException {}
+
+ @Override
+ public void waitSnapshotCompletion() throws InterruptedException {}
+
+ @Override
+ public void streamingPaused() {}
+
+ @Override
+ public void waitStreamingPaused() throws InterruptedException {}
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java
index 58bffa63c8c..d931a65b5af 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java
@@ -17,9 +17,8 @@
package org.apache.flink.cdc.connectors.db2.source.utils;
-import io.debezium.config.Configuration;
import io.debezium.connector.db2.Db2Connection;
-import io.debezium.jdbc.JdbcConfiguration;
+import io.debezium.connector.db2.Db2ConnectorConfig;
import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.RelationalDatabaseConnectorConfig;
import io.debezium.relational.RelationalTableFilters;
@@ -40,8 +39,9 @@ public class Db2ConnectionUtils {
public static Db2Connection createDb2Connection(
RelationalDatabaseConnectorConfig connectorConfig) {
- Configuration dbzConnectorConfig = connectorConfig.getJdbcConfig();
- return new Db2Connection(JdbcConfiguration.adapt(dbzConnectorConfig));
+ // Debezium 2.7 takes the Db2ConnectorConfig itself rather than a JdbcConfiguration; the
+ // connection needs the config to resolve the CDC schema and the Db2 platform adapter.
+ return new Db2Connection((Db2ConnectorConfig) connectorConfig);
}
public static List listTables(JdbcConnection jdbc, RelationalTableFilters tableFilters)
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java
index 34e328dadca..2266ff1e433 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java
@@ -22,19 +22,21 @@
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+import io.debezium.config.CommonConnectorConfig;
import io.debezium.connector.db2.Db2Connection;
import io.debezium.connector.db2.Db2ConnectorConfig;
import io.debezium.connector.db2.Db2DatabaseSchema;
-import io.debezium.connector.db2.Db2TopicSelector;
+import io.debezium.connector.db2.Db2ValueConverters;
import io.debezium.connector.db2.Lsn;
import io.debezium.connector.db2.SourceInfo;
import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.Column;
import io.debezium.relational.Table;
import io.debezium.relational.TableId;
-import io.debezium.schema.TopicSelector;
-import io.debezium.util.SchemaNameAdjuster;
+import io.debezium.schema.SchemaNameAdjuster;
+import io.debezium.spi.topic.TopicNamingStrategy;
import org.apache.kafka.connect.source.SourceRecord;
import javax.annotation.Nullable;
@@ -229,11 +231,24 @@ public static PreparedStatement readTableSplitDataStatement(
public static Db2DatabaseSchema createDb2DatabaseSchema(
Db2ConnectorConfig connectorConfig, Db2Connection connection) {
- TopicSelector topicSelector = Db2TopicSelector.defaultSelector(connectorConfig);
+ TopicNamingStrategy topicSelector;
+ // Debezium 2.0 resolves classes through the thread context class loader; pin it to the
+ // loader that loaded Debezium so a stale/closed Flink user class loader cannot break it.
+ try (TemporaryClassLoaderContext ignored =
+ TemporaryClassLoaderContext.of(CommonConnectorConfig.class.getClassLoader())) {
+ topicSelector =
+ connectorConfig.getTopicNamingStrategy(
+ CommonConnectorConfig.TOPIC_NAMING_STRATEGY);
+ }
SchemaNameAdjuster schemaNameAdjuster = SchemaNameAdjuster.create();
+ // Debezium 2.0 Db2DatabaseSchema requires the value converters explicitly.
+ Db2ValueConverters valueConverters =
+ new Db2ValueConverters(
+ connectorConfig.getDecimalMode(),
+ connectorConfig.getTemporalPrecisionMode());
return new Db2DatabaseSchema(
- connectorConfig, schemaNameAdjuster, topicSelector, connection);
+ connectorConfig, valueConverters, schemaNameAdjuster, topicSelector, connection);
}
// --------------------------private method-------------------------------
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactoryTest.java
new file mode 100644
index 00000000000..278c929fd04
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactoryTest.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.db2.source.config;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+
+import io.debezium.config.CommonConnectorConfig;
+import io.debezium.connector.db2.Db2ConnectorConfig;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link Db2SourceConfigFactory}. */
+class Db2SourceConfigFactoryTest {
+
+ /**
+ * Debezium 2.0 renamed {@code database.server.name} to {@code topic.prefix}. Without it the
+ * connector configuration does not validate and the topic naming strategy cannot be built.
+ */
+ @Test
+ void testTopicPrefixIsConfigured() {
+ Db2SourceConfigFactory factory = new Db2SourceConfigFactory();
+ factory.hostname("localhost");
+ factory.port(50000);
+ factory.username("db2inst1");
+ factory.password("flinkpw");
+ factory.databaseList("testdb");
+ factory.tableList("testdb.DB2INST1.PRODUCTS");
+ factory.startupOptions(StartupOptions.initial());
+
+ Db2ConnectorConfig connectorConfig =
+ new Db2ConnectorConfig(factory.create(0).getDbzConfiguration());
+
+ assertThat(connectorConfig.getLogicalName()).isEqualTo("Db2_transaction_log_source");
+ assertThat(
+ connectorConfig.getTopicNamingStrategy(
+ CommonConnectorConfig.TOPIC_NAMING_STRATEGY))
+ .isNotNull();
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/embedded/EmbeddedEngineChangeEvent.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/embedded/EmbeddedEngineChangeEvent.java
index cf39b79f4a3..731a3ff61f7 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/embedded/EmbeddedEngineChangeEvent.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/embedded/EmbeddedEngineChangeEvent.java
@@ -6,24 +6,31 @@
package io.debezium.embedded;
-import org.apache.flink.cdc.debezium.internal.DebeziumChangeFetcher;
-
import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.Header;
import io.debezium.engine.RecordChangeEvent;
import org.apache.kafka.connect.source.SourceRecord;
+import java.util.List;
+
/**
- * Copied from Debezium project. Make it public to be accessible from {@link DebeziumChangeFetcher}.
+ * Copied from Debezium project(2.7.4.Final)..
+ *
+ *
Change 1: the constructor is public so {@code DebeziumChangeConsumer} can construct the event
+ * when handing records to the legacy DataStream source (upstream keeps it package-private).
*/
-public class EmbeddedEngineChangeEvent implements ChangeEvent, RecordChangeEvent {
+public class EmbeddedEngineChangeEvent implements ChangeEvent, RecordChangeEvent {
private final K key;
private final V value;
+ private final List> headers;
private final SourceRecord sourceRecord;
- public EmbeddedEngineChangeEvent(K key, V value, SourceRecord sourceRecord) {
+ public EmbeddedEngineChangeEvent(
+ K key, V value, List> headers, SourceRecord sourceRecord) {
this.key = key;
this.value = value;
+ this.headers = headers;
this.sourceRecord = sourceRecord;
}
@@ -37,6 +44,12 @@ public V value() {
return value;
}
+ @SuppressWarnings("unchecked")
+ @Override
+ public List> headers() {
+ return headers;
+ }
+
@Override
public V record() {
return value;
@@ -47,6 +60,11 @@ public String destination() {
return sourceRecord.topic();
}
+ @Override
+ public Integer partition() {
+ return sourceRecord.kafkaPartition();
+ }
+
public SourceRecord sourceRecord() {
return sourceRecord;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java
index 18e9ef93ac1..33645185cd9 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java
@@ -25,7 +25,12 @@
import static org.apache.flink.util.Preconditions.checkNotNull;
-/** A bounded cache for table filter results. */
+/**
+ * A bounded cache for table filter results. *
+ *
+ *
This is a Flink CDC owned class (not a Debezium fork), placed in the {@code io.debezium}
+ * package for access to package-private Debezium members.
+ */
public class CachedTableFilter implements TableFilter {
private static final long TABLE_FILTER_CACHE_MAXIMUM_SIZE = 32 * 1024;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/HistorizedRelationalDatabaseConnectorConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/HistorizedRelationalDatabaseConnectorConfig.java
deleted file mode 100644
index 98d25b74f11..00000000000
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/HistorizedRelationalDatabaseConnectorConfig.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/*
- * Copyright Debezium Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.debezium.relational;
-
-import io.debezium.config.ConfigDefinition;
-import io.debezium.config.Configuration;
-import io.debezium.config.Field;
-import io.debezium.relational.Selectors.TableIdToStringMapper;
-import io.debezium.relational.Tables.TableFilter;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryListener;
-import io.debezium.relational.history.DatabaseHistoryMetrics;
-import io.debezium.relational.history.HistoryRecordComparator;
-import io.debezium.relational.history.KafkaDatabaseHistory;
-import org.apache.kafka.common.config.ConfigDef.Importance;
-import org.apache.kafka.common.config.ConfigDef.Type;
-import org.apache.kafka.common.config.ConfigDef.Width;
-import org.apache.kafka.connect.errors.ConnectException;
-import org.apache.kafka.connect.source.SourceConnector;
-
-/**
- * Copied from Debezium project. Configuration options shared across the relational CDC connectors
- * which use a persistent database schema history.
- *
- *
Added JMX_METRICS_ENABLED option.
- */
-public abstract class HistorizedRelationalDatabaseConnectorConfig
- extends RelationalDatabaseConnectorConfig {
-
- protected static final int DEFAULT_SNAPSHOT_FETCH_SIZE = 2_000;
-
- private boolean useCatalogBeforeSchema;
- private final String logicalName;
- private final Class extends SourceConnector> connectorClass;
- private final boolean multiPartitionMode;
-
- /**
- * The database history class is hidden in the {@link #configDef()} since that is designed to
- * work with a user interface, and in these situations using Kafka is the only way to go.
- */
- public static final Field DATABASE_HISTORY =
- Field.create("database.history")
- .withDisplayName("Database history class")
- .withType(Type.CLASS)
- .withWidth(Width.LONG)
- .withImportance(Importance.LOW)
- .withInvisibleRecommender()
- .withDescription(
- "The name of the DatabaseHistory class that should be used to store and recover database schema changes. "
- + "The configuration properties for the history are prefixed with the '"
- + DatabaseHistory.CONFIGURATION_FIELD_PREFIX_STRING
- + "' string.")
- .withDefault(KafkaDatabaseHistory.class.getName());
-
- public static final Field JMX_METRICS_ENABLED =
- Field.create(DatabaseHistory.CONFIGURATION_FIELD_PREFIX_STRING + "metrics.enabled")
- .withDisplayName("Skip DDL statements that cannot be parsed")
- .withType(Type.BOOLEAN)
- .withImportance(Importance.LOW)
- .withDescription("Whether to enable JMX history metrics")
- .withDefault(false);
-
- protected static final ConfigDefinition CONFIG_DEFINITION =
- RelationalDatabaseConnectorConfig.CONFIG_DEFINITION
- .edit()
- .history(
- DATABASE_HISTORY,
- DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS,
- DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL,
- DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL,
- KafkaDatabaseHistory.BOOTSTRAP_SERVERS,
- KafkaDatabaseHistory.TOPIC,
- KafkaDatabaseHistory.RECOVERY_POLL_ATTEMPTS,
- KafkaDatabaseHistory.RECOVERY_POLL_INTERVAL_MS,
- KafkaDatabaseHistory.KAFKA_QUERY_TIMEOUT_MS)
- .create();
-
- protected HistorizedRelationalDatabaseConnectorConfig(
- Class extends SourceConnector> connectorClass,
- Configuration config,
- String logicalName,
- TableFilter systemTablesFilter,
- boolean useCatalogBeforeSchema,
- int defaultSnapshotFetchSize,
- ColumnFilterMode columnFilterMode,
- boolean multiPartitionMode) {
- super(
- config,
- logicalName,
- systemTablesFilter,
- TableId::toString,
- defaultSnapshotFetchSize,
- columnFilterMode);
- this.useCatalogBeforeSchema = useCatalogBeforeSchema;
- this.logicalName = logicalName;
- this.connectorClass = connectorClass;
- this.multiPartitionMode = multiPartitionMode;
- }
-
- protected HistorizedRelationalDatabaseConnectorConfig(
- Class extends SourceConnector> connectorClass,
- Configuration config,
- String logicalName,
- TableFilter systemTablesFilter,
- TableIdToStringMapper tableIdMapper,
- boolean useCatalogBeforeSchema,
- ColumnFilterMode columnFilterMode,
- boolean multiPartitionMode) {
- super(
- config,
- logicalName,
- systemTablesFilter,
- tableIdMapper,
- DEFAULT_SNAPSHOT_FETCH_SIZE,
- columnFilterMode);
- this.useCatalogBeforeSchema = useCatalogBeforeSchema;
- this.logicalName = logicalName;
- this.connectorClass = connectorClass;
- this.multiPartitionMode = multiPartitionMode;
- }
-
- /** Returns a configured (but not yet started) instance of the database history. */
- public DatabaseHistory getDatabaseHistory() {
- Configuration config = getConfig();
-
- DatabaseHistory databaseHistory =
- config.getInstance(
- HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY,
- DatabaseHistory.class);
- if (databaseHistory == null) {
- throw new ConnectException(
- "Unable to instantiate the database history class "
- + config.getString(
- HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY));
- }
-
- // Do not remove the prefix from the subset of config properties ...
- Configuration dbHistoryConfig =
- config.subset(DatabaseHistory.CONFIGURATION_FIELD_PREFIX_STRING, false)
- .edit()
- .withDefault(DatabaseHistory.NAME, getLogicalName() + "-dbhistory")
- .withDefault(
- KafkaDatabaseHistory.INTERNAL_CONNECTOR_CLASS,
- connectorClass.getName())
- .withDefault(KafkaDatabaseHistory.INTERNAL_CONNECTOR_ID, logicalName)
- .build();
-
- DatabaseHistoryListener listener =
- config.getBoolean(JMX_METRICS_ENABLED)
- ? new DatabaseHistoryMetrics(this, multiPartitionMode)
- : DatabaseHistoryListener.NOOP;
-
- HistoryRecordComparator historyComparator = getHistoryRecordComparator();
- databaseHistory.configure(
- dbHistoryConfig, historyComparator, listener, useCatalogBeforeSchema); // validates
-
- return databaseHistory;
- }
-
- public boolean useCatalogBeforeSchema() {
- return useCatalogBeforeSchema;
- }
-
- /**
- * Returns a comparator to be used when recovering records from the schema history, making sure
- * no history entries newer than the offset we resume from are recovered (which could happen
- * when restarting a connector after history records have been persisted but no new offset has
- * been committed yet).
- */
- protected abstract HistoryRecordComparator getHistoryRecordComparator();
-}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java
index 2b0688a5dc4..4fdc3630442 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java
@@ -3,12 +3,10 @@
*
* Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
*/
-
package io.debezium.relational;
import io.debezium.data.Envelope.Operation;
import io.debezium.pipeline.AbstractChangeRecordEmitter;
-import io.debezium.pipeline.spi.ChangeRecordEmitter;
import io.debezium.pipeline.spi.OffsetContext;
import io.debezium.pipeline.spi.Partition;
import io.debezium.schema.DataCollectionSchema;
@@ -22,13 +20,14 @@
import java.util.Optional;
/**
- * Copied from Debezium 1.9.8.Final.
- *
- *
Base class for {@link ChangeRecordEmitter} implementations based on a relational database.
+ * Base class for {@link io.debezium.pipeline.spi.ChangeRecordEmitter} implementations based on a
+ * relational database.
*
- *
This class overrides the emit methods to put some values in the header.
+ *
Copied from Debezium project(2.7.4.Final)..
*
- *
Line 59 ~ 257: add other headers and emit.
+ *
Change 1: add the {@code getEmitConnectHeaders()} hook and pass its value as the change record
+ * headers in every emit method, so subclasses (e.g. the Oracle LogMiner emitter's ROWID header) can
+ * attach Connect headers to emitted records.
*/
public abstract class RelationalChangeRecordEmitter
extends AbstractChangeRecordEmitter
{
@@ -39,8 +38,12 @@ public abstract class RelationalChangeRecordEmitter
public static final String PK_UPDATE_OLDKEY_FIELD = "__debezium.oldkey";
public static final String PK_UPDATE_NEWKEY_FIELD = "__debezium.newkey";
- public RelationalChangeRecordEmitter(P partition, OffsetContext offsetContext, Clock clock) {
- super(partition, offsetContext, clock);
+ public RelationalChangeRecordEmitter(
+ P partition,
+ OffsetContext offsetContext,
+ Clock clock,
+ RelationalDatabaseConnectorConfig connectorConfig) {
+ super(partition, offsetContext, clock, connectorConfig);
}
@Override
@@ -87,7 +90,7 @@ protected void emitCreateRecord(Receiver
receiver, TableSchema tableSchema)
if (skipEmptyMessages() && (newColumnValues == null || newColumnValues.length == 0)) {
// This case can be hit on UPDATE / DELETE when there's no primary key defined while
// using certain decoders
- LOGGER.warn(
+ LOGGER.debug(
"no new values found for table '{}' from create message at '{}'; skipping record",
tableSchema,
getOffset().getSourceInfo());
@@ -140,12 +143,25 @@ protected void emitUpdateRecord(Receiver
receiver, TableSchema tableSchema)
Struct oldValue = tableSchema.valueFromColumnData(oldColumnValues);
if (skipEmptyMessages() && (newColumnValues == null || newColumnValues.length == 0)) {
- LOGGER.warn(
+ LOGGER.debug(
"no new values found for table '{}' from update message at '{}'; skipping record",
tableSchema,
getOffset().getSourceInfo());
return;
}
+
+ /*
+ * If skip.messages.without.change is configured true,
+ * Skip Publishing the message in case there is no change in monitored columns
+ * (Postgres) Only works if REPLICA IDENTITY is set to FULL - as oldValues won't be available
+ */
+ if (skipMessagesWithoutChange() && Objects.nonNull(newValue) && newValue.equals(oldValue)) {
+ LOGGER.debug(
+ "No new values found for table '{}' in included columns from update message at '{}'; skipping record",
+ tableSchema,
+ getOffset().getSourceInfo());
+ return;
+ }
// some configurations does not provide old values in case of updates
// in this case we handle all updates as regular ones
if (oldKey == null || Objects.equals(oldKey, newKey)) {
@@ -181,7 +197,7 @@ protected void emitDeleteRecord(Receiver
receiver, TableSchema tableSchema)
Struct oldValue = tableSchema.valueFromColumnData(oldColumnValues);
if (skipEmptyMessages() && (oldColumnValues == null || oldColumnValues.length == 0)) {
- LOGGER.warn(
+ LOGGER.debug(
"no old values found for table '{}' from delete message at '{}'; skipping record",
tableSchema,
getOffset().getSourceInfo());
@@ -219,8 +235,8 @@ protected void emitTruncateRecord(Receiver
receiver, TableSchema schema)
/**
* Whether empty data messages should be ignored.
*
- * @return true if empty data messages coming from data source should be ignored. Typical use
- * case are PostgreSQL changes without FULL replica identity.
+ * @return true if empty data messages coming from data source should be ignored. Typical
+ * use case are PostgreSQL changes without FULL replica identity.
*/
protected boolean skipEmptyMessages() {
return false;
@@ -234,45 +250,34 @@ protected void emitUpdateAsPrimaryKeyChangeRecord(
Struct oldValue,
Struct newValue)
throws InterruptedException {
- ConnectHeaders headers = getEmitConnectHeaders().orElse(new ConnectHeaders());
+ final OffsetContext offset = getOffset();
+ final Struct sourceInfo = offset.getSourceInfo();
+
+ ConnectHeaders headers = new ConnectHeaders();
headers.add(PK_UPDATE_NEWKEY_FIELD, newKey, tableSchema.keySchema());
Struct envelope =
tableSchema
.getEnvelopeSchema()
- .delete(
- oldValue,
- getOffset().getSourceInfo(),
- getClock().currentTimeAsInstant());
+ .delete(oldValue, sourceInfo, getClock().currentTimeAsInstant());
receiver.changeRecord(
- getPartition(),
- tableSchema,
- Operation.DELETE,
- oldKey,
- envelope,
- getOffset(),
- headers);
+ getPartition(), tableSchema, Operation.DELETE, oldKey, envelope, offset, headers);
- headers = getEmitConnectHeaders().orElse(new ConnectHeaders());
+ headers = new ConnectHeaders();
headers.add(PK_UPDATE_OLDKEY_FIELD, oldKey, tableSchema.keySchema());
envelope =
tableSchema
.getEnvelopeSchema()
- .create(
- newValue,
- getOffset().getSourceInfo(),
- getClock().currentTimeAsInstant());
+ .create(newValue, sourceInfo, getClock().currentTimeAsInstant());
receiver.changeRecord(
- getPartition(),
- tableSchema,
- Operation.CREATE,
- newKey,
- envelope,
- getOffset(),
- headers);
+ getPartition(), tableSchema, Operation.CREATE, newKey, envelope, offset, headers);
}
+ /**
+ * Returns the Connect headers to attach to emitted change records; {@link Optional#empty()} by
+ * default. Flink CDC addition — see the class javadoc.
+ */
protected Optional getEmitConnectHeaders() {
return Optional.empty();
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
index 4d91ff83c2e..98e0aeb008b 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
@@ -9,30 +9,32 @@
import io.debezium.relational.Selectors.TableIdToStringMapper;
import io.debezium.relational.Selectors.TableSelectionPredicateBuilder;
import io.debezium.relational.Tables.TableFilter;
-import io.debezium.relational.history.DatabaseHistory;
+import io.debezium.relational.history.SchemaHistory;
import io.debezium.schema.DataCollectionFilters;
import java.util.function.Predicate;
-import static io.debezium.relational.RelationalDatabaseConnectorConfig.COLUMN_BLACKLIST;
import static io.debezium.relational.RelationalDatabaseConnectorConfig.COLUMN_EXCLUDE_LIST;
/**
- * Copied from Debezium 1.9.8.Final.
+ * Copied from Debezium project(2.7.4.Final)..
*
- *
Line 98: cache table filter results.
+ *
Change 1: wrap the table filter in {@link CachedTableFilter} so repeated filter evaluations
+ * (once per change record) hit a cache instead of re-running the predicate chain.
*
- *
Line 148: add a method to update the tableFilter variable.
+ *
Change 2: add {@code setDataCollectionFilters} so Flink CDC can replace the table filter after
+ * construction (used by the MySQL source config when the captured-table list is refined).
*/
public class RelationalTableFilters implements DataCollectionFilters {
- // Filter that filters tables based only on datbase/schema/system table filters but not table
+ // Filter that filters tables based only on database/schema/system table filters but not table
// filters
// Represents the list of tables whose schema needs to be captured
private final TableFilter eligibleTableFilter;
// Filter that filters tables based on table filters
private TableFilter tableFilter;
private final Predicate databaseFilter;
+ private final Predicate schemaFilter;
private final String excludeColumns;
/**
@@ -45,26 +47,23 @@ public class RelationalTableFilters implements DataCollectionFilters {
public RelationalTableFilters(
Configuration config,
TableFilter systemTablesFilter,
- TableIdToStringMapper tableIdMapper) {
+ TableIdToStringMapper tableIdMapper,
+ boolean useCatalogBeforeSchema) {
// Define the filter that provides the list of tables that could be captured if configured
final TableSelectionPredicateBuilder eligibleTables =
Selectors.tableSelector()
.includeDatabases(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.DATABASE_INCLUDE_LIST,
- RelationalDatabaseConnectorConfig.DATABASE_WHITELIST))
+ config.getString(
+ RelationalDatabaseConnectorConfig.DATABASE_INCLUDE_LIST))
.excludeDatabases(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.DATABASE_EXCLUDE_LIST,
- RelationalDatabaseConnectorConfig.DATABASE_BLACKLIST))
+ config.getString(
+ RelationalDatabaseConnectorConfig.DATABASE_EXCLUDE_LIST))
.includeSchemas(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.SCHEMA_INCLUDE_LIST,
- RelationalDatabaseConnectorConfig.SCHEMA_WHITELIST))
+ config.getString(
+ RelationalDatabaseConnectorConfig.SCHEMA_INCLUDE_LIST))
.excludeSchemas(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.SCHEMA_EXCLUDE_LIST,
- RelationalDatabaseConnectorConfig.SCHEMA_BLACKLIST));
+ config.getString(
+ RelationalDatabaseConnectorConfig.SCHEMA_EXCLUDE_LIST));
final Predicate eligibleTablePredicate = eligibleTables.build();
Predicate finalEligibleTablePredicate =
@@ -78,14 +77,12 @@ public RelationalTableFilters(
Predicate tablePredicate =
eligibleTables
.includeTables(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.TABLE_INCLUDE_LIST,
- RelationalDatabaseConnectorConfig.TABLE_WHITELIST),
+ config.getString(
+ RelationalDatabaseConnectorConfig.TABLE_INCLUDE_LIST),
tableIdMapper)
.excludeTables(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.TABLE_EXCLUDE_LIST,
- RelationalDatabaseConnectorConfig.TABLE_BLACKLIST),
+ config.getString(
+ RelationalDatabaseConnectorConfig.TABLE_EXCLUDE_LIST),
tableIdMapper)
.build();
@@ -93,21 +90,41 @@ public RelationalTableFilters(
config.getBoolean(RelationalDatabaseConnectorConfig.TABLE_IGNORE_BUILTIN)
? tablePredicate.and(systemTablesFilter::isIncluded)
: tablePredicate;
-
- TableFilter initialTableFilter = finalTablePredicate::test;
- this.tableFilter = CachedTableFilter.from(initialTableFilter);
+ String signalDataCollection =
+ config.getString(RelationalDatabaseConnectorConfig.SIGNAL_DATA_COLLECTION);
+ if (signalDataCollection != null) {
+ TableId signalDataCollectionTableId =
+ TableId.parse(signalDataCollection, useCatalogBeforeSchema);
+ if (!finalTablePredicate.test(signalDataCollectionTableId)) {
+ final Predicate signalDataCollectionPredicate =
+ Selectors.tableSelector()
+ .includeTables(
+ tableIdMapper.toString(signalDataCollectionTableId),
+ tableIdMapper)
+ .build();
+ finalTablePredicate = finalTablePredicate.or(signalDataCollectionPredicate);
+ }
+ }
+ this.tableFilter = CachedTableFilter.from(finalTablePredicate::test);
// Define the database filter using the include and exclude lists for database names ...
this.databaseFilter =
Selectors.databaseSelector()
.includeDatabases(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.DATABASE_INCLUDE_LIST,
- RelationalDatabaseConnectorConfig.DATABASE_WHITELIST))
+ config.getString(
+ RelationalDatabaseConnectorConfig.DATABASE_INCLUDE_LIST))
.excludeDatabases(
- config.getFallbackStringProperty(
- RelationalDatabaseConnectorConfig.DATABASE_EXCLUDE_LIST,
- RelationalDatabaseConnectorConfig.DATABASE_BLACKLIST))
+ config.getString(
+ RelationalDatabaseConnectorConfig.DATABASE_EXCLUDE_LIST))
+ .build();
+ this.schemaFilter =
+ Selectors.databaseSelector()
+ .includeDatabases(
+ config.getString(
+ RelationalDatabaseConnectorConfig.SCHEMA_INCLUDE_LIST))
+ .excludeDatabases(
+ config.getString(
+ RelationalDatabaseConnectorConfig.SCHEMA_EXCLUDE_LIST))
.build();
Predicate eligibleSchemaPredicate =
@@ -115,13 +132,15 @@ public RelationalTableFilters(
? systemTablesFilter::isIncluded
: x -> true;
- this.schemaSnapshotFilter =
- config.getBoolean(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL)
- ? eligibleSchemaPredicate.and(initialTableFilter::isIncluded)::test
- : eligibleSchemaPredicate::test;
+ if (config.getBoolean(SchemaHistory.STORE_ONLY_CAPTURED_TABLES_DDL)) {
+ this.schemaSnapshotFilter = eligibleSchemaPredicate.and(tableFilter::isIncluded)::test;
+ } else if (config.getBoolean(SchemaHistory.STORE_ONLY_CAPTURED_DATABASES_DDL)) {
+ this.schemaSnapshotFilter = finalEligibleTablePredicate::test;
+ } else {
+ this.schemaSnapshotFilter = eligibleSchemaPredicate::test;
+ }
- this.excludeColumns =
- config.getFallbackStringProperty(COLUMN_EXCLUDE_LIST, COLUMN_BLACKLIST);
+ this.excludeColumns = config.getString(COLUMN_EXCLUDE_LIST);
}
@Override
@@ -141,10 +160,15 @@ public Predicate databaseFilter() {
return databaseFilter;
}
+ public Predicate schemaFilter() {
+ return schemaFilter;
+ }
+
public String getExcludeColumns() {
return excludeColumns;
}
+ /** Replaces the table filter. Flink CDC addition — see the class javadoc. */
public void setDataCollectionFilters(TableFilter tableFilter) {
this.tableFilter = tableFilter;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/DebeziumSourceFunction.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/DebeziumSourceFunction.java
index 21d822a06ee..2302242bb58 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/DebeziumSourceFunction.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/DebeziumSourceFunction.java
@@ -49,6 +49,7 @@
import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.debezium.DebeziumException;
import io.debezium.document.DocumentReader;
import io.debezium.document.DocumentWriter;
import io.debezium.embedded.Connect;
@@ -370,6 +371,16 @@ private void snapshotHistoryRecordsState() throws Exception {
@Override
public void run(SourceContext sourceContext) throws Exception {
properties.putIfAbsent("name", "engine");
+ // Debezium 2.x tags the JMX ObjectName of a multi-partition connector's metrics with
+ // "task.id" in addition to the topic prefix. Kafka Connect sets that property; Flink
+ // never has, so Metrics#metricName hands null to Sanitizer.jmxSanitize and the task
+ // fails with a NullPointerException before producing anything. SQL Server is the only
+ // multi-partition connector today, but the property is connector-agnostic, so set it
+ // here for every embedded engine. The subtask index is the natural analogue of a
+ // Connect task id and keeps the name unique within the TaskManager JVM.
+ properties.putIfAbsent(
+ "task.id",
+ String.valueOf(getRuntimeContext().getTaskInfo().getIndexOfThisSubtask()));
properties.setProperty("offset.storage", FlinkOffsetBackingStore.class.getCanonicalName());
if (restoredOffsetState != null) {
// restored from state
@@ -381,6 +392,13 @@ public void run(SourceContext sourceContext) throws Exception {
properties.putIfAbsent("offset.flush.interval.ms", String.valueOf(Long.MAX_VALUE));
// disable tombstones
properties.setProperty("tombstones.on.delete", "false");
+ // Debezium 2.4 made the embedded engine restart the connector itself on a retriable
+ // error, retrying forever by default ("errors.max.retries" = -1). Debezium 2.3 and older
+ // had no such loop: the error reached the completion callback, the Flink job failed and
+ // Flink's restart strategy recovered it from the last checkpoint. Keep that behavior - an
+ // engine silently retrying forever looks like a source that simply stopped producing.
+ // Users who want the Debezium-side retries can still set "debezium.errors.max.retries".
+ properties.putIfAbsent("errors.max.retries", "0");
if (engineInstanceName == null) {
// not restore from recovery
engineInstanceName = UUID.randomUUID().toString();
@@ -393,7 +411,7 @@ public void run(SourceContext sourceContext) throws Exception {
// see
// https://stackoverflow.com/questions/57147584/debezium-error-schema-isnt-know-to-this-connector
// and https://debezium.io/blog/2018/03/16/note-on-database-history-topic-configuration/
- properties.setProperty("database.history", determineDatabase().getCanonicalName());
+ properties.setProperty("schema.history.internal", determineDatabase().getCanonicalName());
// we have to filter out the heartbeat events, otherwise the deserializer will fail
String dbzHeartbeatPrefix =
@@ -420,7 +438,16 @@ public void run(SourceContext sourceContext) throws Exception {
// Close the handover and prepare to exit.
handover.close();
} else {
- handover.reportError(error);
+ // Debezium may report a failure without a throwable, e.g.
+ // when the connector configuration does not validate. In
+ // that case the message is the only diagnostic we have.
+ handover.reportError(
+ error != null
+ ? error
+ : new DebeziumException(
+ message == null
+ ? "The Debezium engine has stopped unexpectedly."
+ : message));
}
})
.build();
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/history/FlinkJsonTableChangeSerializer.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/history/FlinkJsonTableChangeSerializer.java
index 234673c3d5c..45e5136c646 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/history/FlinkJsonTableChangeSerializer.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/history/FlinkJsonTableChangeSerializer.java
@@ -58,8 +58,10 @@ public Document toDocument(TableChange tableChange) {
document.setString("type", tableChange.getType().name());
document.setString("id", tableChange.getId().toDoubleQuotedString());
- document.setDocument("table", toDocument(tableChange.getTable()));
- document.setString("comment", tableChange.getTable().comment());
+ if (tableChange.getTable() != null) {
+ document.setDocument("table", toDocument(tableChange.getTable()));
+ document.setString("comment", tableChange.getTable().comment());
+ }
return document;
}
@@ -127,7 +129,7 @@ public TableChanges deserialize(Array array, boolean useCatalogBeforeSchema) {
} else if (change.getType() == TableChangeType.ALTER) {
tableChanges.alter(change.getTable());
} else if (change.getType() == TableChangeType.DROP) {
- tableChanges.drop(change.getTable());
+ tableChanges.drop(change.getId());
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/DebeziumChangeConsumer.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/DebeziumChangeConsumer.java
index 03aae4b3fe8..25ffbb51ff0 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/DebeziumChangeConsumer.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/DebeziumChangeConsumer.java
@@ -28,6 +28,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -78,8 +79,9 @@ public void commitOffset(DebeziumOffset offset) throws InterruptedException {
"DUMMY",
Schema.BOOLEAN_SCHEMA,
true);
- EmbeddedEngineChangeEvent changeEvent =
- new EmbeddedEngineChangeEvent<>(null, recordWrapper, recordWrapper);
+ EmbeddedEngineChangeEvent changeEvent =
+ new EmbeddedEngineChangeEvent<>(
+ null, recordWrapper, Collections.emptyList(), recordWrapper);
currentCommitter.markProcessed(changeEvent);
currentCommitter.markBatchFinished();
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseHistory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseHistory.java
index f4f880b9f39..983887966c1 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseHistory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseHistory.java
@@ -18,11 +18,11 @@
package org.apache.flink.cdc.debezium.internal;
import io.debezium.config.Configuration;
-import io.debezium.relational.history.AbstractDatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryException;
-import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.AbstractSchemaHistory;
import io.debezium.relational.history.HistoryRecord;
import io.debezium.relational.history.HistoryRecordComparator;
+import io.debezium.relational.history.SchemaHistoryException;
+import io.debezium.relational.history.SchemaHistoryListener;
import java.util.Collection;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -40,9 +40,10 @@
* records will be stored in state (grow infinitely). We may need to come up with a
* FileSystemDatabaseHistory in the future to store history in HDFS.
*/
-public class FlinkDatabaseHistory extends AbstractDatabaseHistory {
+public class FlinkDatabaseHistory extends AbstractSchemaHistory {
- public static final String DATABASE_HISTORY_INSTANCE_NAME = "database.history.instance.name";
+ public static final String DATABASE_HISTORY_INSTANCE_NAME =
+ "schema.history.internal.instance.name";
private ConcurrentLinkedQueue schemaRecords;
private String instanceName;
@@ -57,7 +58,7 @@ private ConcurrentLinkedQueue getRegisteredHistoryRecord(String in
public void configure(
Configuration config,
HistoryRecordComparator comparator,
- DatabaseHistoryListener listener,
+ SchemaHistoryListener listener,
boolean useCatalogBeforeSchema) {
super.configure(config, comparator, listener, useCatalogBeforeSchema);
this.instanceName = config.getString(DATABASE_HISTORY_INSTANCE_NAME);
@@ -68,6 +69,16 @@ public void configure(
registerHistory(instanceName, schemaRecords);
}
+ // Debezium 2.0 wires SchemaHistoryMetrics in as the schema history listener, and
+ // AbstractSchemaHistory#start does nothing but hand it a started() callback. That callback
+ // registers a JMX MBean named only after the connector context and the topic prefix, so
+ // every parallel subtask sharing a TaskManager JVM asks for the very same name. Debezium
+ // answers a name clash by sleeping 5 seconds and retrying, twelve times. Flink CDC
+ // publishes its own metrics and never reads these MBeans, and Debezium 1.9 did not
+ // register them either, so skip the registration.
+ @Override
+ public void start() {}
+
@Override
public void stop() {
super.stop();
@@ -75,7 +86,7 @@ public void stop() {
}
@Override
- protected void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
+ protected void storeRecord(HistoryRecord record) throws SchemaHistoryException {
this.schemaRecords.add(new SchemaRecord(record));
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseSchemaHistory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseSchemaHistory.java
index b9fc8fd7795..d509d60029f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseSchemaHistory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkDatabaseSchemaHistory.java
@@ -23,14 +23,15 @@
import io.debezium.relational.TableId;
import io.debezium.relational.Tables;
import io.debezium.relational.ddl.DdlParser;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryException;
-import io.debezium.relational.history.DatabaseHistoryListener;
import io.debezium.relational.history.HistoryRecord;
import io.debezium.relational.history.HistoryRecordComparator;
+import io.debezium.relational.history.SchemaHistory;
+import io.debezium.relational.history.SchemaHistoryException;
+import io.debezium.relational.history.SchemaHistoryListener;
import io.debezium.relational.history.TableChanges;
import io.debezium.schema.DatabaseSchema;
+import java.time.Instant;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -51,16 +52,17 @@
* FlinkDatabaseHistory}. Because it only maintains the latest schema of the table rather than all
* history DDLs, it's useful to prevent OOM when meet massive history DDLs.
*/
-public class FlinkDatabaseSchemaHistory implements DatabaseHistory {
+public class FlinkDatabaseSchemaHistory implements SchemaHistory {
- public static final String DATABASE_HISTORY_INSTANCE_NAME = "database.history.instance.name";
+ public static final String DATABASE_HISTORY_INSTANCE_NAME =
+ "schema.history.internal.instance.name";
private final FlinkJsonTableChangeSerializer tableChangesSerializer =
new FlinkJsonTableChangeSerializer();
private ConcurrentMap latestTables;
private String instanceName;
- private DatabaseHistoryListener listener;
+ private SchemaHistoryListener listener;
private boolean storeOnlyMonitoredTablesDdl;
private boolean skipUnparseableDDL;
private boolean useCatalogBeforeSchema;
@@ -69,11 +71,11 @@ public class FlinkDatabaseSchemaHistory implements DatabaseHistory {
public void configure(
Configuration config,
HistoryRecordComparator comparator,
- DatabaseHistoryListener listener,
+ SchemaHistoryListener listener,
boolean useCatalogBeforeSchema) {
this.instanceName = config.getString(DATABASE_HISTORY_INSTANCE_NAME);
this.listener = listener;
- this.storeOnlyMonitoredTablesDdl = config.getBoolean(STORE_ONLY_MONITORED_TABLES_DDL);
+ this.storeOnlyMonitoredTablesDdl = config.getBoolean(STORE_ONLY_CAPTURED_TABLES_DDL);
this.skipUnparseableDDL = config.getBoolean(SKIP_UNPARSEABLE_DDL_STATEMENTS);
this.useCatalogBeforeSchema = useCatalogBeforeSchema;
@@ -91,14 +93,19 @@ public void configure(
}
@Override
- public void start() {
- listener.started();
- }
+ // Debezium 2.0 wires SchemaHistoryMetrics in as the schema history listener. Its
+ // started() callback registers a JMX MBean whose name is built only from the connector
+ // context and the topic prefix, so every parallel subtask sharing a TaskManager JVM asks
+ // for the very same name. Debezium answers a name clash by sleeping 5 seconds and
+ // retrying, twelve times, so every reader but the first stalls for up to a minute each
+ // time it opens a split. Flink CDC publishes its own metrics and never reads these
+ // MBeans, and Debezium 1.9 did not register them either, so skip the registration.
+ public void start() {}
@Override
public void record(
Map source, Map position, String databaseName, String ddl)
- throws DatabaseHistoryException {
+ throws SchemaHistoryException {
throw new UnsupportedOperationException(
String.format(
"The %s cannot work with 'debezium.internal.implementation' = 'legacy',"
@@ -114,8 +121,9 @@ public void record(
String databaseName,
String schemaName,
String ddl,
- TableChanges changes)
- throws DatabaseHistoryException {
+ TableChanges changes,
+ Instant timestamp)
+ throws SchemaHistoryException {
for (TableChanges.TableChange change : changes) {
switch (change.getType()) {
case CREATE:
@@ -134,7 +142,8 @@ public void record(
}
}
listener.onChangeApplied(
- new HistoryRecord(source, position, databaseName, schemaName, ddl, changes));
+ new HistoryRecord(
+ source, position, databaseName, schemaName, ddl, changes, timestamp));
}
@Override
@@ -186,12 +195,12 @@ public void initializeStorage() {
// do nothing
}
- @Override
+ // Debezium 2.2 moved storeOnlyCapturedTables()/skipUnparseableDdlStatements() from the
+ // SchemaHistory interface to HistorizedDatabaseSchema; kept here as plain helpers.
public boolean storeOnlyCapturedTables() {
return storeOnlyMonitoredTablesDdl;
}
- @Override
public boolean skipUnparseableDdlStatements() {
return skipUnparseableDDL;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkOffsetBackingStore.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkOffsetBackingStore.java
index dbce64eb75f..7281b9f3754 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkOffsetBackingStore.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/internal/FlinkOffsetBackingStore.java
@@ -36,8 +36,10 @@
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -90,10 +92,14 @@ public void configure(WorkerConfig config) {
String engineName = (String) conf.get(EmbeddedEngine.ENGINE_NAME.name());
Converter keyConverter = new JsonConverter();
Converter valueConverter = new JsonConverter();
- keyConverter.configure(config.originals(), true);
- Map valueConfigs = new HashMap<>(conf);
- valueConfigs.put("schemas.enable", false);
- valueConverter.configure(valueConfigs, true);
+ // Debezium 2.0's embedded engine serializes offset keys/values as raw JSON without the
+ // {"schema":...,"payload":...} envelope. Disable schemas on BOTH converters so the key we
+ // seed here matches the key the engine looks up on recovery (otherwise the restored offset
+ // is never found and the connector falls back to a schema-only-recovery snapshot).
+ Map converterConfigs = new HashMap<>(conf);
+ converterConfigs.put("schemas.enable", false);
+ keyConverter.configure(converterConfigs, true);
+ valueConverter.configure(converterConfigs, false);
OffsetStorageWriter offsetWriter =
new OffsetStorageWriter(
this,
@@ -200,4 +206,14 @@ public Future set(
return null;
});
}
+
+ /**
+ * Added in Kafka Connect 3.5 (KAFKA-14304) for exactly-once source support and the offset-reset
+ * REST endpoint. Neither code path is exercised by the Flink embedded engine, which drives
+ * offsets through Flink state, so an empty set is a safe no-op.
+ */
+ @Override
+ public Set