[mysql] Ignore exceptions when encountered unparseable DDL

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));

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}.
  */