diff --git a/flink-connector-mysql-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/mysql/MySQLSource.java b/flink-connector-mysql-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/mysql/MySQLSource.java index 0b8514f99..c41d89bae 100644 --- a/flink-connector-mysql-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/mysql/MySQLSource.java +++ b/flink-connector-mysql-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/mysql/MySQLSource.java @@ -151,6 +151,7 @@ public class MySQLSource { 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)); if (serverId != null) { props.setProperty("database.server.id", String.valueOf(serverId)); diff --git a/flink-connector-postgres-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java b/flink-connector-postgres-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java index d420f7ed6..f18a26907 100644 --- a/flink-connector-postgres-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java +++ b/flink-connector-postgres-cdc/src/main/java/com/alibaba/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java @@ -18,9 +18,6 @@ package com.alibaba.ververica.cdc.connectors.postgres.table; -import static com.alibaba.ververica.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; -import static com.alibaba.ververica.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; - import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.ReadableConfig; @@ -34,6 +31,9 @@ import org.apache.flink.table.utils.TableSchemaUtils; import java.util.HashSet; import java.util.Set; +import static com.alibaba.ververica.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; +import static com.alibaba.ververica.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; + /** * Factory for creating configured instance of {@link PostgreSQLTableSource}. */