[feature][connector][mysql-cdc] add MySQL CDC enumerator (#3481)

* [feature][connector][mysql-cdc] add mysql-cdc enumerator classes

* [chore] cdc-base package rename to org.apache.seatunnel

* [chore] reference static method

* [feature][connector][mysql-cdc] support earliest startup mode

* [feature][connector][mysql-cdc] add MySqlIncrementalSource

* fix error

Author: ashulin

seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/ConsumerWithException.java renamed to seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/function/ConsumerWithException.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.engine.common.utils;
+package org.apache.seatunnel.common.utils.function;
 
 @FunctionalInterface
 public interface ConsumerWithException<T>  {

seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/function/FunctionWithException.java

@@ -0,0 +1,37 @@
+/*
+ * 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.seatunnel.common.utils.function;
+
+/**
+ * A functional interface for a {@link java.util.function.Function} that may throw exceptions.
+ *
+ * @param <T> The type of the argument to the function.
+ * @param <R> The type of the result of the supplier.
+ * @param <E> The type of Exceptions thrown by this function.
+ */
+@FunctionalInterface
+public interface FunctionWithException<T, R, E extends Throwable>  {
+    /**
+     * Applies this function to the given argument.
+     *
+     * @param value The argument to the function.
+     * @return The result of thus supplier.
+     * @throws E This function may throw an exception.
+     */
+    R apply(T value) throws E;
+}

seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/RunnableWithException.java renamed to seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/function/RunnableWithException.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.engine.common.utils;
+package org.apache.seatunnel.common.utils.function;
 
 /**
  * Similar to a {@link Runnable}, this interface is used to capture a block of code to be executed.

seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/SupplierWithException.java renamed to seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/function/SupplierWithException.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.engine.common.utils;
+package org.apache.seatunnel.common.utils.function;
 
 /**
  * A functional interface for a {@link java.util.function.Supplier} that may throw exceptions.

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/seatunnel/connectors/cdc/base/config/BaseSourceConfig.java renamed to seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/BaseSourceConfig.java

@@ -15,11 +15,12 @@
  * limitations under the License.
  */
 
-package org.seatunnel.connectors.cdc.base.config;
+package org.apache.seatunnel.connectors.cdc.base.config;
+
+import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
 
 import io.debezium.config.Configuration;
 import lombok.Getter;
-import org.seatunnel.connectors.cdc.base.source.IncrementalSource;
 
 import java.util.Properties;
 

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/seatunnel/connectors/cdc/base/config/JdbcSourceConfig.java renamed to seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/JdbcSourceConfig.java

@@ -15,10 +15,11 @@
  * limitations under the License.
  */
 
-package org.seatunnel.connectors.cdc.base.config;
+package org.apache.seatunnel.connectors.cdc.base.config;
+
+import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
 
 import io.debezium.relational.RelationalDatabaseConnectorConfig;
-import org.seatunnel.connectors.cdc.base.source.IncrementalSource;
 
 import java.time.Duration;
 import java.util.List;

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

@@ -15,10 +15,10 @@
  * limitations under the License.
  */
 
-package org.seatunnel.connectors.cdc.base.config;
+package org.apache.seatunnel.connectors.cdc.base.config;
 
-import org.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
-import org.seatunnel.connectors.cdc.base.option.SourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
 
 import java.time.Duration;
 import java.util.Arrays;

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/seatunnel/connectors/cdc/base/config/SourceConfig.java renamed to seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/SourceConfig.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.seatunnel.connectors.cdc.base.config;
+package org.apache.seatunnel.connectors.cdc.base.config;
 
 import java.io.Serializable;
 

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/seatunnel/connectors/cdc/base/config/StartupConfig.java renamed to seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java

@@ -15,14 +15,15 @@
  * limitations under the License.
  */
 
-package org.seatunnel.connectors.cdc.base.config;
+package org.apache.seatunnel.connectors.cdc.base.config;
+
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
 
 import lombok.AllArgsConstructor;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
-import org.seatunnel.connectors.cdc.base.option.StartupMode;
-import org.seatunnel.connectors.cdc.base.source.offset.Offset;
-import org.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
 
 import java.io.Serializable;
 
@@ -45,7 +46,7 @@ public Offset getStartupOffset(OffsetFactory offsetFactory) {
             case INITIAL:
                 return null;
             case TIMESTAMP:
-                return offsetFactory.timstamp(timestamp);
+                return offsetFactory.timestamp(timestamp);
             default:
                 throw new IllegalArgumentException(String.format("The %s mode is not supported.", startupMode));
         }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/seatunnel/connectors/cdc/base/config/StopConfig.java renamed to seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StopConfig.java

@@ -15,14 +15,15 @@
  * limitations under the License.
  */
 
-package org.seatunnel.connectors.cdc.base.config;
+package org.apache.seatunnel.connectors.cdc.base.config;
+
+import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
 
 import lombok.AllArgsConstructor;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
-import org.seatunnel.connectors.cdc.base.option.StopMode;
-import org.seatunnel.connectors.cdc.base.source.offset.Offset;
-import org.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
 
 import java.io.Serializable;
 
@@ -46,7 +47,7 @@ public Offset getStopOffset(OffsetFactory offsetFactory) {
             case SPECIFIC:
                 return offsetFactory.specific(specificOffsetFile, specificOffsetPos);
             case TIMESTAMP:
-                return offsetFactory.timstamp(timestamp);
+                return offsetFactory.timestamp(timestamp);
             default:
                 throw new IllegalArgumentException(String.format("The %s mode is not supported.", stopMode));
         }