[FLINK-36605] Bump debezium to last 2.x version

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/pom.xml

@@ -187,6 +187,7 @@ limitations under the License.
                                     <include>io.debezium:debezium-core</include>
                                     <include>io.debezium:debezium-ddl-parser</include>
                                     <include>io.debezium:debezium-connector-mysql</include>
+                                    <include>io.debezium:debezium-connector-binlog</include>
                                     <include>org.apache.flink:flink-connector-debezium</include>
                                     <include>org.apache.flink:flink-connector-mysql-cdc</include>
                                     <include>org.antlr:antlr4-runtime</include>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessor.java

@@ -41,7 +41,14 @@ public class MySqlMetadataAccessor implements MetadataAccessor {
     public MySqlMetadataAccessor(MySqlSourceConfig sourceConfig) {
         this.sourceConfig = sourceConfig;
         this.partition =
-                new MySqlPartition(sourceConfig.getMySqlConnectorConfig().getLogicalName());
+                new MySqlPartition(
+                        sourceConfig.getMySqlConnectorConfig().getLogicalName(),
+                        sourceConfig
+                                .getMySqlConnectorConfig()
+                                .getConfig()
+                                .getString(
+                                        io.debezium.relational.RelationalDatabaseConnectorConfig
+                                                .DATABASE_NAME));
     }
 
     /**

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java

@@ -96,7 +96,8 @@ public void exitCopyCreateTable(MySqlParser.CopyCreateTableContext ctx) {
                             tableId,
                             original.columns(),
                             original.primaryKeyColumnNames(),
-                            original.defaultCharsetName());
+                            original.defaultCharsetName(),
+                            Collections.emptyList());
             parser.signalCreateTable(tableId, ctx);
             Schema.Builder builder = Schema.newBuilder();
             original.columns().forEach(column -> builder.column(toCdcColumn(column)));
@@ -170,7 +171,9 @@ public void exitColumnCreateTable(MySqlParser.ColumnCreateTableContext ctx) {
     public void enterColumnDeclaration(MySqlParser.ColumnDeclarationContext ctx) {
         parser.runIfNotNull(
                 () -> {
-                    String columnName = parser.parseName(ctx.uid());
+                    // Debezium 2.1 changed the MySQL grammar: columnDeclaration now exposes the
+                    // column name via fullColumnName() instead of uid().
+                    String columnName = parser.parseName(ctx.fullColumnName().uid());
                     ColumnEditor columnEditor = Column.editor().name(columnName);
                     if (columnDefinitionListener == null) {
                         columnDefinitionListener =

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParser.java

@@ -22,6 +22,7 @@
 import io.debezium.antlr.AntlrDdlParserListener;
 import io.debezium.antlr.DataTypeResolver;
 import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
+import io.debezium.connector.mysql.charset.MySqlCharsetRegistry;
 import io.debezium.ddl.parser.mysql.generated.MySqlParser;
 import io.debezium.relational.Tables;
 
@@ -40,7 +41,17 @@ public class CustomMySqlAntlrDdlParser extends MySqlAntlrDdlParser {
 
     public CustomMySqlAntlrDdlParser(
             boolean includeComments, boolean tinyInt1isBit, boolean isTableIdCaseInsensitive) {
-        super(true, false, includeComments, null, Tables.TableFilter.includeAll());
+        // Debezium 2.7 added the BinlogCharsetRegistry argument. It must not be null: the
+        // parser dereferences it in extractCharset() whenever a COLLATE clause appears without
+        // an accompanying CHARACTER SET, which is how MySQL renders a collation. This is the
+        // same implementation Debezium's own MySqlDatabaseSchema resolves from its registry.
+        super(
+                true,
+                false,
+                includeComments,
+                null,
+                Tables.TableFilter.includeAll(),
+                new MySqlCharsetRegistry());
         this.parsedEvents = new LinkedList<>();
         this.tinyInt1isBit = tinyInt1isBit;
         this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java

@@ -37,6 +37,7 @@
 import org.apache.flink.connector.base.source.reader.RecordEmitter;
 
 import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
+import io.debezium.connector.mysql.charset.MySqlCharsetRegistry;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.Column;
 import io.debezium.relational.RelationalDatabaseConnectorConfig;
@@ -315,7 +316,14 @@ private synchronized MySqlAntlrDdlParser getParser() {
                                     false);
             mySqlAntlrDdlParser =
                     new MySqlAntlrDdlParser(
-                            true, false, includeComments, null, Tables.TableFilter.includeAll());
+                            true,
+                            false,
+                            includeComments,
+                            null,
+                            Tables.TableFilter.includeAll(),
+                            // Debezium 2.7 added the BinlogCharsetRegistry argument; it is
+                            // dereferenced when parsing a COLLATE clause, so it must not be null.
+                            new MySqlCharsetRegistry());
         }
         return mySqlAntlrDdlParser;
     }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlSchemaUtils.java

@@ -23,8 +23,8 @@
 import org.apache.flink.cdc.connectors.mysql.schema.MySqlSchema;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
 
-import io.debezium.connector.mysql.MySqlConnection;
 import io.debezium.connector.mysql.MySqlPartition;
+import io.debezium.connector.mysql.jdbc.MySqlConnection;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.Table;
 import io.debezium.relational.history.TableChanges;

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MysqlPipelineNewlyAddedTableITCase.java

@@ -55,7 +55,7 @@
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.util.ExceptionUtils;
 
-import io.debezium.connector.mysql.MySqlConnection;
+import io.debezium.connector.mysql.jdbc.MySqlConnection;
 import io.debezium.jdbc.JdbcConnection;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.connectors.oracle.source;
 
+import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.cdc.common.configuration.Configuration;
@@ -146,7 +147,7 @@ public void testAlterAddAllColumnTypeStatement() throws Exception {
         dbzProperties.setProperty("database.connection.adapter", "logminer");
         dbzProperties.setProperty("log.mining.strategy", "online_catalog");
         dbzProperties.setProperty("snapshot.locking.mode", "none");
-        dbzProperties.setProperty("database.history.store.only.captured.tables.ddl", "true");
+        dbzProperties.setProperty("schema.history.internal.store.only.captured.tables.ddl", "true");
         dbzProperties.setProperty("include.schema.changes", "true");
         OracleSourceConfigFactory configFactory = new OracleSourceConfigFactory();
         configFactory.username(CONNECTOR_USER);
@@ -605,7 +606,7 @@ public void testParseAlterStatement() throws Exception {
         dbzProperties.setProperty("database.connection.adapter", "logminer");
         dbzProperties.setProperty("log.mining.strategy", "online_catalog");
         dbzProperties.setProperty("snapshot.locking.mode", "none");
-        dbzProperties.setProperty("database.history.store.only.captured.tables.ddl", "true");
+        dbzProperties.setProperty("schema.history.internal.store.only.captured.tables.ddl", "true");
         dbzProperties.setProperty("include.schema.changes", "true");
         OracleSourceConfigFactory configFactory = new OracleSourceConfigFactory();
         configFactory.username(CONNECTOR_USER);
@@ -732,7 +733,7 @@ public void testInitialStartupMode() throws Exception {
         dbzProperties.setProperty("database.connection.adapter", "logminer");
         dbzProperties.setProperty("log.mining.strategy", "online_catalog");
         dbzProperties.setProperty("snapshot.locking.mode", "none");
-        dbzProperties.setProperty("database.history.store.only.captured.tables.ddl", "true");
+        dbzProperties.setProperty("schema.history.internal.store.only.captured.tables.ddl", "true");
         dbzProperties.setProperty("include.schema.changes", "true");
         OracleSourceConfigFactory configFactory = new OracleSourceConfigFactory();
         configFactory
@@ -1272,7 +1273,7 @@ public void testSnapshotOnlyMode() throws Exception {
         dbzProperties.setProperty("database.connection.adapter", "logminer");
         dbzProperties.setProperty("log.mining.strategy", "online_catalog");
         dbzProperties.setProperty("snapshot.locking.mode", "none");
-        dbzProperties.setProperty("database.history.store.only.captured.tables.ddl", "true");
+        dbzProperties.setProperty("schema.history.internal.store.only.captured.tables.ddl", "true");
         dbzProperties.setProperty("include.schema.changes", "true");
         OracleSourceConfigFactory configFactory = new OracleSourceConfigFactory();
         configFactory.username(CONNECTOR_USER);
@@ -1368,7 +1369,7 @@ private void restartFromSavePoint(
         dbzProperties.setProperty("database.connection.adapter", "logminer");
         dbzProperties.setProperty("log.mining.strategy", "online_catalog");
         dbzProperties.setProperty("snapshot.locking.mode", "none");
-        dbzProperties.setProperty("database.history.store.only.captured.tables.ddl", "true");
+        dbzProperties.setProperty("schema.history.internal.store.only.captured.tables.ddl", "true");
         dbzProperties.setProperty("include.schema.changes", "true");
 
         configFactory.username(CONNECTOR_USER);
@@ -1432,15 +1433,22 @@ private void restartFromSavePoint(
             statement.execute(String.format("DELETE FROM %s.PRODUCTS WHERE ID = 111 ", "DEBEZIUM"));
         }
 
-        Thread.sleep(30_000);
+        // A snapshot-only source is BOUNDED, so the job terminates as soon as the snapshot is
+        // done. Sleeping a fixed 30s lets it reach FINISHED first, and stopWithSavepoint can
+        // never succeed on a terminated job. Wait only until the job is RUNNING so the savepoint
+        // is taken while the snapshot is still in flight. Unbounded modes keep running either
+        // way, so they retain the original settle time.
+        if ("snapshot".equals(options.get(SCAN_STARTUP_MODE.key()))) {
+            awaitJobRunning(jobClient);
+        } else {
+            Thread.sleep(30_000);
+        }
 
         // Trigger a savepoint and cancel the job
         LOG.info("Triggering savepoint");
         finishedSavePointPath = triggerSavepointWithRetry(jobClient, savepointDirectory);
         LOG.info("Savepoint created at: {}", finishedSavePointPath);
-        if (!jobClient.getJobStatus().get().name().equals("FINISHED")) {
-            jobClient.cancel().get();
-        }
+        cancelIfRunning(jobClient);
         iterator.close();
 
         // Restore from savepoint
@@ -1505,7 +1513,7 @@ private void restartFromSavePoint(
             restoreAfterEvents.add(restoredIterator.next());
         }
         restoredIterator.close();
-        restoredJobClient.cancel().get();
+        cancelIfRunning(restoredJobClient);
         // Check if CreateTableEvent for new_products is present
         boolean hasCreateTableEvent =
                 restoreAfterEvents.stream().anyMatch(event -> event instanceof CreateTableEvent);
@@ -1518,6 +1526,39 @@ private void restartFromSavePoint(
         env.close();
     }
 
+    /**
+     * Cancels the job unless it has already reached a terminal state. A snapshot-only source is
+     * bounded, so the job can finish on its own before the test reaches the cancel, and {@code
+     * cancel()} on a finished job throws {@code FlinkJobTerminatedWithoutCancellationException}.
+     */
+    private void cancelIfRunning(JobClient jobClient) throws Exception {
+        if (!jobClient.getJobStatus().get().isTerminalState()) {
+            jobClient.cancel().get();
+        }
+    }
+
+    /**
+     * Waits until the job is RUNNING so that a savepoint can still be taken, failing fast with a
+     * meaningful message if it reaches a terminal state first.
+     */
+    private void awaitJobRunning(JobClient jobClient) throws Exception {
+        final long deadline = System.currentTimeMillis() + 60_000;
+        while (System.currentTimeMillis() < deadline) {
+            JobStatus status = jobClient.getJobStatus().get();
+            if (status == JobStatus.RUNNING) {
+                return;
+            }
+            if (status.isTerminalState()) {
+                throw new IllegalStateException(
+                        "Job reached terminal state "
+                                + status
+                                + " before a savepoint could be taken");
+            }
+            Thread.sleep(100);
+        }
+        throw new IllegalStateException("Job did not reach RUNNING within 60s");
+    }
+
     // Helper method to trigger a savepoint with retry mechanism
     private String triggerSavepointWithRetry(JobClient jobClient, String savepointDirectory)
             throws Exception {
@@ -1529,6 +1570,13 @@ private String triggerSavepointWithRetry(JobClient jobClient, String savepointDi
                         .stopWithSavepoint(true, savepointDirectory, SavepointFormatType.DEFAULT)
                         .get();
             } catch (Exception e) {
+                // A terminated job can never accept a savepoint, so retrying just burns the
+                // whole budget and then reports a misleading timeout instead of the real cause.
+                JobStatus status = jobClient.getJobStatus().get();
+                if (status.isTerminalState()) {
+                    throw new IllegalStateException(
+                            "Cannot stop with savepoint: job already reached " + status, e);
+                }
                 retryCount++;
                 LOG.error(
                         "Retry {}/{}: Failed to trigger savepoint: {}",

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/PostgresSchemaUtils.java

@@ -23,15 +23,15 @@
 import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
 import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
 
+import io.debezium.config.CommonConnectorConfig;
 import io.debezium.connector.postgresql.PostgresConnectorConfig;
 import io.debezium.connector.postgresql.PostgresObjectUtils;
 import io.debezium.connector.postgresql.PostgresSchema;
-import io.debezium.connector.postgresql.PostgresTopicSelector;
 import io.debezium.connector.postgresql.TypeRegistry;
 import io.debezium.connector.postgresql.connection.PostgresConnection;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.Table;
-import io.debezium.schema.TopicSelector;
+import io.debezium.spi.topic.TopicNamingStrategy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -155,8 +155,10 @@ public static Schema getTableSchema(
             PostgresConnection jdbc) {
         try {
             // fetch table schemas
-            TopicSelector<io.debezium.relational.TableId> topicSelector =
-                    PostgresTopicSelector.create(sourceConfig.getDbzConnectorConfig());
+            TopicNamingStrategy<io.debezium.relational.TableId> topicSelector =
+                    sourceConfig
+                            .getDbzConnectorConfig()
+                            .getTopicNamingStrategy(CommonConnectorConfig.TOPIC_NAMING_STRATEGY);
             PostgresConnection.PostgresValueConverterBuilder valueConverterBuilder =
                     newPostgresValueConverterBuilder(sourceConfig.getDbzConnectorConfig());
             PostgresSchema postgresSchema =

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerEventDeserializerTest.java

@@ -213,7 +213,7 @@ private static SourceRecord buildSchemaChangeRecord(TableChangeType type, List<C
                 tableChanges.alter(table);
                 break;
             case DROP:
-                tableChanges.drop(table);
+                tableChanges.drop(table.id());
                 break;
             default:
                 throw new IllegalArgumentException("Unsupported type " + type);
@@ -226,7 +226,8 @@ private static SourceRecord buildSchemaChangeRecord(TableChangeType type, List<C
                         DBZ_TABLE_ID.catalog(),
                         DBZ_TABLE_ID.schema(),
                         "ddl",
-                        tableChanges);
+                        tableChanges,
+                        java.time.Instant.now());
 
         String historyJson = DOCUMENT_WRITER.write(historyRecord.document());