[WIP][Common] The abstraction of Incremental Snapshot Framework

flink-cdc-base/pom.xml

@@ -0,0 +1,137 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>flink-cdc-connectors</artifactId>
+        <groupId>com.ververica</groupId>
+        <version>2.2-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>flink-cdc-base</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.ververica</groupId>
+            <artifactId>flink-connector-debezium</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <artifactId>kafka-log4j-appender</artifactId>
+                    <groupId>org.apache.kafka</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP</artifactId>
+            <version>4.0.3</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-runtime-blink_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-test-utils</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-tests</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+
+        <!-- Debezium test dependencies -->
+
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-mysql</artifactId>
+            <version>${debezium.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+
+        <!-- test dependencies on TestContainers -->
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>mysql</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/config/BaseSourceConfig.java

@@ -0,0 +1,95 @@
+/*
+ * 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 com.ververica.cdc.connectors.base.config;
+
+import com.ververica.cdc.connectors.base.options.StartupOptions;
+import com.ververica.cdc.connectors.base.source.JdbcIncrementalSource;
+import io.debezium.config.Configuration;
+
+import java.util.Properties;
+
+/** A basic Source configuration which is used by {@link JdbcIncrementalSource}. */
+public abstract class BaseSourceConfig implements SourceConfig {
+
+    private static final long serialVersionUID = 1L;
+
+    protected final StartupOptions startupOptions;
+    protected final int splitSize;
+    protected final int splitMetaGroupSize;
+    protected final double distributionFactorUpper;
+    protected final double distributionFactorLower;
+    protected final boolean includeSchemaChanges;
+
+    // --------------------------------------------------------------------------------------------
+    // Debezium Configurations
+    // --------------------------------------------------------------------------------------------
+    protected final Properties dbzProperties;
+    protected transient Configuration dbzConfiguration;
+
+    public BaseSourceConfig(
+            StartupOptions startupOptions,
+            int splitSize,
+            int splitMetaGroupSize,
+            double distributionFactorUpper,
+            double distributionFactorLower,
+            boolean includeSchemaChanges,
+            Properties dbzProperties,
+            Configuration dbzConfiguration) {
+        this.startupOptions = startupOptions;
+        this.splitSize = splitSize;
+        this.splitMetaGroupSize = splitMetaGroupSize;
+        this.distributionFactorUpper = distributionFactorUpper;
+        this.distributionFactorLower = distributionFactorLower;
+        this.includeSchemaChanges = includeSchemaChanges;
+        this.dbzProperties = dbzProperties;
+        this.dbzConfiguration = dbzConfiguration;
+    }
+
+    public StartupOptions getStartupOptions() {
+        return startupOptions;
+    }
+
+    public int getSplitSize() {
+        return splitSize;
+    }
+
+    public int getSplitMetaGroupSize() {
+        return splitMetaGroupSize;
+    }
+
+    public double getDistributionFactorUpper() {
+        return distributionFactorUpper;
+    }
+
+    public double getDistributionFactorLower() {
+        return distributionFactorLower;
+    }
+
+    public boolean isIncludeSchemaChanges() {
+        return includeSchemaChanges;
+    }
+
+    public Properties getDbzProperties() {
+        return dbzProperties;
+    }
+
+    public Configuration getDbzConfiguration() {
+        return Configuration.from(dbzProperties);
+    }
+}

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/config/JdbcSourceConfig.java

@@ -0,0 +1,142 @@
+/*
+ * 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 com.ververica.cdc.connectors.base.config;
+
+import com.ververica.cdc.connectors.base.options.StartupOptions;
+import com.ververica.cdc.connectors.base.source.JdbcIncrementalSource;
+import io.debezium.config.Configuration;
+import io.debezium.relational.RelationalDatabaseConnectorConfig;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * A Source configuration which is used by {@link JdbcIncrementalSource} which used JDBC data
+ * source.
+ */
+public abstract class JdbcSourceConfig extends BaseSourceConfig {
+
+    protected final String driverClassName;
+    protected final String hostname;
+    protected final int port;
+    protected final String username;
+    protected final String password;
+    protected final List<String> databaseList;
+    protected final List<String> tableList;
+    protected final int fetchSize;
+    protected final String serverTimeZone;
+    protected final Duration connectTimeout;
+    protected final int connectMaxRetries;
+    protected final int connectionPoolSize;
+
+    public JdbcSourceConfig(
+            StartupOptions startupOptions,
+            List<String> databaseList,
+            List<String> tableList,
+            int splitSize,
+            int splitMetaGroupSize,
+            double distributionFactorUpper,
+            double distributionFactorLower,
+            boolean includeSchemaChanges,
+            Properties dbzProperties,
+            Configuration dbzConfiguration,
+            String driverClassName,
+            String hostname,
+            int port,
+            String username,
+            String password,
+            int fetchSize,
+            String serverTimeZone,
+            Duration connectTimeout,
+            int connectMaxRetries,
+            int connectionPoolSize) {
+        super(
+                startupOptions,
+                splitSize,
+                splitMetaGroupSize,
+                distributionFactorUpper,
+                distributionFactorLower,
+                includeSchemaChanges,
+                dbzProperties,
+                dbzConfiguration);
+        this.driverClassName = driverClassName;
+        this.hostname = hostname;
+        this.port = port;
+        this.username = username;
+        this.password = password;
+        this.databaseList = databaseList;
+        this.tableList = tableList;
+        this.fetchSize = fetchSize;
+        this.serverTimeZone = serverTimeZone;
+        this.connectTimeout = connectTimeout;
+        this.connectMaxRetries = connectMaxRetries;
+        this.connectionPoolSize = connectionPoolSize;
+    }
+
+    public abstract RelationalDatabaseConnectorConfig getDbzConnectorConfig();
+
+    public String getDriverClassName() {
+        return driverClassName;
+    }
+
+    public String getHostname() {
+        return hostname;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public List<String> getDatabaseList() {
+        return databaseList;
+    }
+
+    public List<String> getTableList() {
+        return tableList;
+    }
+
+    public int getFetchSize() {
+        return fetchSize;
+    }
+
+    public String getServerTimeZone() {
+        return serverTimeZone;
+    }
+
+    public Duration getConnectTimeout() {
+        return connectTimeout;
+    }
+
+    public int getConnectMaxRetries() {
+        return connectMaxRetries;
+    }
+
+    public int getConnectionPoolSize() {
+        return connectionPoolSize;
+    }
+}