[feature][cdc] Fixed error in mysql cdc under real-time job (#3666)

* [feature][cdc] Fixed error in mysql cdc under real-time job

* [chore] license header

Author: ashulin

plugin-mapping.properties

@@ -156,4 +156,5 @@ seatunnel.source.Jira = connector-http-jira
 seatunnel.source.Gitlab = connector-http-gitlab
 seatunnel.sink.RabbitMQ = connector-rabbitmq
 seatunnel.source.RabbitMQ = connector-rabbitmq
-seatunnel.source.OpenMldb = connector-openmldb
+seatunnel.source.OpenMldb = connector-openmldb
+seatunnel.source.MySQL-CDC = connector-cdc-mysql

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/io/debezium/relational/TableId.java

@@ -0,0 +1,280 @@
+/*
+ * 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 io.debezium.relational;
+
+import io.debezium.annotation.Immutable;
+import io.debezium.relational.Selectors.TableIdToStringMapper;
+import io.debezium.schema.DataCollectionId;
+
+import java.io.Serializable;
+
+/**
+ * Unique identifier for a database table.
+ *
+ */
+@Immutable
+public final class TableId implements DataCollectionId, Comparable<TableId>, Serializable {
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * Parse the supplied string, extracting up to the first 3 parts into a TableID.
+     *
+     * @param str the string representation of the table identifier; may not be null
+     * @return the table ID, or null if it could not be parsed
+     */
+    public static TableId parse(String str) {
+        return parse(str, true);
+    }
+
+    /**
+     * Parse the supplied string, extracting up to the first 3 parts into a TableID.
+     *
+     * @param str the string representation of the table identifier; may not be null
+     * @param useCatalogBeforeSchema {@code true} if the parsed string contains only 2 items and the first should be used as
+     *            the catalog and the second as the table name, or {@code false} if the first should be used as the schema and the
+     *            second as the table name
+     * @return the table ID, or null if it could not be parsed
+     */
+    public static TableId parse(String str, boolean useCatalogBeforeSchema) {
+        String[] parts = TableIdParser.parse(str).toArray(new String[0]);
+
+        return TableId.parse(parts, parts.length, useCatalogBeforeSchema);
+    }
+
+    /**
+     * Parse the supplied string, extracting up to the first 3 parts into a TableID.
+     *
+     * @param parts the parts of the identifier; may not be null
+     * @param numParts the number of parts to use for the table identifier
+     * @param useCatalogBeforeSchema {@code true} if the parsed string contains only 2 items and the first should be used as
+     *            the catalog and the second as the table name, or {@code false} if the first should be used as the schema and the
+     *            second as the table name
+     * @return the table ID, or null if it could not be parsed
+     */
+    protected static TableId parse(String[] parts, int numParts, boolean useCatalogBeforeSchema) {
+        if (numParts == 0) {
+            return null;
+        }
+        if (numParts == 1) {
+            return new TableId(null, null, parts[0]); // table only
+        }
+        if (numParts == 2) {
+            if (useCatalogBeforeSchema) {
+                return new TableId(parts[0], null, parts[1]); // catalog & table only
+            }
+            return new TableId(null, parts[0], parts[1]); // schema & table only
+        }
+        return new TableId(parts[0], parts[1], parts[2]); // catalog, schema & table
+    }
+
+    private final String catalogName;
+    private final String schemaName;
+    private final String tableName;
+    private final String id;
+
+    /**
+     * Create a new table identifier.
+     *
+     * @param catalogName the name of the database catalog that contains the table; may be null if the JDBC driver does not
+     *            show a schema for this table
+     * @param schemaName the name of the database schema that contains the table; may be null if the JDBC driver does not
+     *            show a schema for this table
+     * @param tableName the name of the table; may not be null
+     * @param tableIdMapper the customization of fully quailified table name
+     */
+    public TableId(String catalogName, String schemaName, String tableName, TableIdToStringMapper tableIdMapper) {
+        this.catalogName = catalogName;
+        this.schemaName = schemaName;
+        this.tableName = tableName;
+        assert this.tableName != null;
+        this.id = tableIdMapper == null ? tableId(this.catalogName, this.schemaName, this.tableName) : tableIdMapper.toString(this);
+    }
+
+    /**
+     * Create a new table identifier.
+     *
+     * @param catalogName the name of the database catalog that contains the table; may be null if the JDBC driver does not
+     *            show a schema for this table
+     * @param schemaName the name of the database schema that contains the table; may be null if the JDBC driver does not
+     *            show a schema for this table
+     * @param tableName the name of the table; may not be null
+     */
+    public TableId(String catalogName, String schemaName, String tableName) {
+        this(catalogName, schemaName, tableName, null);
+    }
+
+    /**
+     * Get the name of the JDBC catalog.
+     *
+     * @return the catalog name, or null if the table does not belong to a catalog
+     */
+    public String catalog() {
+        return catalogName;
+    }
+
+    /**
+     * Get the name of the JDBC schema.
+     *
+     * @return the JDBC schema name, or null if the table does not belong to a JDBC schema
+     */
+    public String schema() {
+        return schemaName;
+    }
+
+    /**
+     * Get the name of the table.
+     *
+     * @return the table name; never null
+     */
+    public String table() {
+        return tableName;
+    }
+
+    @Override
+    public String identifier() {
+        return id;
+    }
+
+    @Override
+    public int compareTo(TableId that) {
+        if (this == that) {
+            return 0;
+        }
+        return this.id.compareTo(that.id);
+    }
+
+    public int compareToIgnoreCase(TableId that) {
+        if (this == that) {
+            return 0;
+        }
+        return this.id.compareToIgnoreCase(that.id);
+    }
+
+    @Override
+    public int hashCode() {
+        return id.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj instanceof TableId) {
+            return this.compareTo((TableId) obj) == 0;
+        }
+        return false;
+    }
+
+    @Override
+    public String toString() {
+        return identifier();
+    }
+
+    /**
+     * Returns a dot-separated String representation of this identifier, quoting all
+     * name parts with the {@code "} char.
+     */
+    public String toDoubleQuotedString() {
+        return toQuotedString('"');
+    }
+
+    /**
+     * Returns a new {@link TableId} with all parts of the identifier using {@code "} character.
+     */
+    public TableId toDoubleQuoted() {
+        return toQuoted('"');
+    }
+
+    /**
+     * Returns a new {@link TableId} that has all parts of the identifier quoted.
+     *
+     * @param quotingChar the character to be used to quote the identifier parts.
+     */
+    public TableId toQuoted(char quotingChar) {
+        String catalogName = null;
+        if (this.catalogName != null && !this.catalogName.isEmpty()) {
+            catalogName = quote(this.catalogName, quotingChar);
+        }
+
+        String schemaName = null;
+        if (this.schemaName != null && !this.schemaName.isEmpty()) {
+            schemaName = quote(this.schemaName, quotingChar);
+        }
+
+        return new TableId(catalogName, schemaName, quote(this.tableName, quotingChar));
+    }
+
+    /**
+     * Returns a dot-separated String representation of this identifier, quoting all
+     * name parts with the given quoting char.
+     */
+    public String toQuotedString(char quotingChar) {
+        StringBuilder quoted = new StringBuilder();
+
+        if (catalogName != null && !catalogName.isEmpty()) {
+            quoted.append(quote(catalogName, quotingChar)).append(".");
+        }
+
+        if (schemaName != null && !schemaName.isEmpty()) {
+            quoted.append(quote(schemaName, quotingChar)).append(".");
+        }
+
+        quoted.append(quote(tableName, quotingChar));
+
+        return quoted.toString();
+    }
+
+    private static String tableId(String catalog, String schema, String table) {
+        if (catalog == null || catalog.length() == 0) {
+            if (schema == null || schema.length() == 0) {
+                return table;
+            }
+            return schema + "." + table;
+        }
+        if (schema == null || schema.length() == 0) {
+            return catalog + "." + table;
+        }
+        return catalog + "." + schema + "." + table;
+    }
+
+    /**
+     * Quotes the given identifier part, e.g. schema or table name.
+     */
+    private static String quote(String identifierPart, char quotingChar) {
+        if (identifierPart == null) {
+            return null;
+        }
+
+        if (identifierPart.isEmpty()) {
+            return new StringBuilder().append(quotingChar).append(quotingChar).toString();
+        }
+
+        if (identifierPart.charAt(0) != quotingChar && identifierPart.charAt(identifierPart.length() - 1) != quotingChar) {
+            identifierPart = identifierPart.replace(quotingChar + "", repeat(quotingChar));
+            identifierPart = quotingChar + identifierPart + quotingChar;
+        }
+
+        return identifierPart;
+    }
+
+    private static String repeat(char quotingChar) {
+        return new StringBuilder().append(quotingChar).append(quotingChar).toString();
+    }
+
+    public TableId toLowercase() {
+        return new TableId(catalogName, schemaName, tableName.toLowerCase());
+    }
+}

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/JdbcSourceConfigFactory.java

@@ -179,13 +179,21 @@ public JdbcSourceConfigFactory startupOptions(StartupConfig startupConfig) {
         return this;
     }
 
+    /** Specifies the stop options. */
+    public JdbcSourceConfigFactory stopOptions(StopConfig stopConfig) {
+        this.stopConfig = stopConfig;
+        return this;
+    }
+
     public JdbcSourceConfigFactory fromReadonlyConfig(ReadonlyConfig config) {
         this.port = config.get(JdbcSourceOptions.PORT);
         this.hostname = config.get(JdbcSourceOptions.HOSTNAME);
+        this.username = config.get(JdbcSourceOptions.USERNAME);
         this.password = config.get(JdbcSourceOptions.PASSWORD);
         // TODO: support multi-table
         this.databaseList = Collections.singletonList(config.get(JdbcSourceOptions.DATABASE_NAME));
-        this.tableList = Collections.singletonList(config.get(JdbcSourceOptions.TABLE_NAME));
+        this.tableList = Collections.singletonList(config.get(JdbcSourceOptions.DATABASE_NAME)
+            + "." + config.get(JdbcSourceOptions.TABLE_NAME));
         this.distributionFactorUpper = config.get(JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
         this.distributionFactorLower = config.get(JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
         this.splitSize = config.get(SourceOptions.SNAPSHOT_SPLIT_SIZE);

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssigner.java

@@ -111,7 +111,7 @@ public Optional<SourceSplitBase> getNext() {
             // we need to wait snapshot-assigner to be completed before
             // assigning the incremental split. Otherwise, records emitted from incremental split
             // might be out-of-order in terms of same primary key with snapshot splits.
-            return snapshotSplitAssigner.getNext();
+            return incrementalSplitAssigner.getNext();
         }
         // no more splits for the assigner
         return Optional.empty();

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java

@@ -51,12 +51,14 @@ public class IncrementalSourceEnumerator
      */
     private final TreeSet<Integer> readersAwaitingSplit;
 
+    private volatile boolean running;
     public IncrementalSourceEnumerator(
             SourceSplitEnumerator.Context<SourceSplitBase> context,
             SplitAssigner splitAssigner) {
         this.context = context;
         this.splitAssigner = splitAssigner;
         this.readersAwaitingSplit = new TreeSet<>();
+        this.running = false;
     }
 
     @Override
@@ -66,7 +68,8 @@ public void open() {
 
     @Override
     public void run() throws Exception {
-
+        this.running = true;
+        assignSplits();
     }
 
     @Override
@@ -77,7 +80,9 @@ public void handleSplitRequest(int subtaskId) {
         }
 
         readersAwaitingSplit.add(subtaskId);
-        assignSplits();
+        if (running) {
+            assignSplits();
+        }
     }
 
     @Override

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java

@@ -124,6 +124,8 @@ public boolean waitingForCompletedSplits() {
     @Override
     public void onCompletedSplits(List<SnapshotSplitWatermark> completedSplitWatermarks) {
         // do nothing
+        completedSplitWatermarks.forEach(watermark ->
+            context.getSplitCompletedOffsets().put(watermark.getSplitId(), watermark.getHighWatermark()));
     }
 
     @Override
@@ -205,7 +207,8 @@ private IncrementalSplit createIncrementalSplit(List<TableId> capturedTables, in
         }
         for (TableId tableId : capturedTables) {
             Offset watermark = tableWatermarks.get(tableId);
-            if (minOffset == null || watermark.isBefore(minOffset)) {
+            if (minOffset == null ||
+                (watermark != null && watermark.isBefore(minOffset))) {
                 minOffset = watermark;
             }
         }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssigner.java

@@ -148,12 +148,16 @@ public void open() {
 
     @Override
     public Optional<SourceSplitBase> getNext() {
+        if (chunkSplitter == null) {
+            return Optional.empty();
+        }
         if (!remainingSplits.isEmpty()) {
             // return remaining splits firstly
             Iterator<SnapshotSplit> iterator = remainingSplits.iterator();
             SnapshotSplit split = iterator.next();
             iterator.remove();
             assignedSplits.put(split.splitId(), split);
+            context.getAssignedSnapshotSplit().put(split.splitId(), split);
             return Optional.of(split);
         } else {
             // it's turn for new table