[Feature][CDC] support mysql cdc

chunjun-connectors/chunjun-connector-jdbc-base/pom.xml

@@ -15,7 +15,7 @@
 	<properties>
 		<!-- TODO 升级4.0？-->
 		<vertx.version>3.9.7</vertx.version>
-		<druid.version>1.2.6</druid.version>
+		<druid.version>1.2.11</druid.version>
 	</properties>
 
 	<dependencies>
@@ -43,6 +43,16 @@
 			<groupId>com.alibaba</groupId>
 			<artifactId>druid</artifactId>
 			<version>${druid.version}</version>
+			<exclusions>
+				<exclusion>
+					<artifactId>jconsole</artifactId>
+					<groupId>com.sun</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>tools</artifactId>
+					<groupId>com.sun</groupId>
+				</exclusion>
+			</exclusions>
 		</dependency>
 	</dependencies>
 </project>

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/conf/CdcConf.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.dtstack.chunjun.connector.jdbc.conf;
+
+import com.dtstack.chunjun.conf.ChunJunCommonConf;
+
+import java.io.Serializable;
+import java.util.List;
+
+public class CdcConf extends ChunJunCommonConf implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    protected String host;
+    protected int port;
+    protected List<String> databaseList;
+    protected List<String> tableList;
+
+    protected String username;
+
+    protected String password;
+
+    protected int serverId;
+
+    public String getHost() {
+        return host;
+    }
+
+    public void setHost(String host) {
+        this.host = host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setPort(int port) {
+        this.port = port;
+    }
+
+    public List<String> getDatabaseList() {
+        return databaseList;
+    }
+
+    public void setDatabaseList(List<String> databaseList) {
+        this.databaseList = databaseList;
+    }
+
+    public List<String> getTableList() {
+        return tableList;
+    }
+
+    public void setTableList(List<String> tableList) {
+        this.tableList = tableList;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public int getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(int serverId) {
+        this.serverId = serverId;
+    }
+}

chunjun-connectors/chunjun-connector-kafka/src/test/java/com/dtstack/chunjun/connector/kafka/sink/KafkaSinkFactoryTest.java

@@ -28,7 +28,7 @@
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.table.data.RowData;
 
-import com.sun.tools.javac.util.Assert;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -107,6 +107,6 @@ public void KafkaSourceFactoryTest() {
         DataStream<RowData> source = new KafkaSourceFactory(config, env).createSource();
         factory = new KafkaSinkFactory(config);
 
-        Assert.checkNonNull(factory.createSink(source));
+        Assert.assertNotNull(factory.createSink(source));
     }
 }

chunjun-connectors/chunjun-connector-kafka/src/test/java/com/dtstack/chunjun/connector/kafka/source/KafkaSourceFactoryTest.java

@@ -25,7 +25,7 @@
 
 import org.apache.flink.configuration.Configuration;
 
-import com.sun.tools.javac.util.Assert;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -92,11 +92,11 @@ public void KafkaSourceFactoryTest() {
                 Main.parseConf(
                         String.format(job, "  \"mode\": \"earliest-offset\""), new Options());
         factory = new KafkaSourceFactory(config, env);
-        Assert.checkNonNull(factory.createSource());
+        Assert.assertNotNull(factory.createSource());
 
         config = Main.parseConf(String.format(job, "  \"mode\": \"latest-offset\""), new Options());
         factory = new KafkaSourceFactory(config, env);
-        Assert.checkNonNull(factory.createSource());
+        Assert.assertNotNull(factory.createSource());
 
         config =
                 Main.parseConf(
@@ -106,7 +106,7 @@ public void KafkaSourceFactoryTest() {
                                         + System.currentTimeMillis()),
                         new Options());
         factory = new KafkaSourceFactory(config, env);
-        Assert.checkNonNull(factory.createSource());
+        Assert.assertNotNull(factory.createSource());
 
         config =
                 Main.parseConf(
@@ -115,6 +115,6 @@ public void KafkaSourceFactoryTest() {
                                 " \"mode\": \"specific-offsets\",\"offset\":\"partition:0,offset:42;partition:1,offset:300\""),
                         new Options());
         factory = new KafkaSourceFactory(config, env);
-        Assert.checkNonNull(factory.createSource());
+        Assert.assertNotNull(factory.createSource());
     }
 }

chunjun-connectors/chunjun-connector-mysqlcdc/pom.xml

@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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>chunjun-connectors</artifactId>
+		<groupId>com.dtstack.chunjun</groupId>
+		<version>1.12-SNAPSHOT</version>
+	</parent>
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>chunjun-connector-mysqlcdc</artifactId>
+	<name>ChunJun : Connectors : MySQLCDC</name>
+
+	<properties>
+		<maven.compiler.source>8</maven.compiler.source>
+		<maven.compiler.target>8</maven.compiler.target>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>com.dtstack.chunjun</groupId>
+			<artifactId>chunjun-connector-jdbc-base</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>mysql</groupId>
+			<artifactId>mysql-connector-java</artifactId>
+			<version>5.1.46</version>
+		</dependency>
+		<dependency>
+			<groupId>com.alibaba.ververica</groupId>
+			<artifactId>flink-connector-mysql-cdc</artifactId>
+			<version>${flinkcdc.version}</version>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-shade-plugin</artifactId>
+				<version>3.2.1</version>
+				<executions>
+					<execution>
+						<phase>package</phase>
+						<goals>
+							<goal>shade</goal>
+						</goals>
+						<configuration>
+							<createDependencyReducedPom>false</createDependencyReducedPom>
+							<artifactSet>
+								<excludes>
+									<exclude>org.slf4j:slf4j-api</exclude>
+									<exclude>log4j:log4j</exclude>
+									<exclude>ch.qos.logback:*</exclude>
+								</excludes>
+							</artifactSet>
+							<relocations>
+								<relocation>
+									<pattern>com.google.common</pattern>
+									<shadedPattern>shade.core.com.google.common</shadedPattern>
+								</relocation>
+							</relocations>
+							<filters>
+								<filter>
+									<artifact>*:*</artifact>
+									<excludes>
+										<exclude>META-INF/*.SF</exclude>
+										<exclude>META-INF/*.DSA</exclude>
+										<exclude>META-INF/*.RSA</exclude>
+									</excludes>
+								</filter>
+							</filters>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+
+			<plugin>
+				<artifactId>maven-antrun-plugin</artifactId>
+				<executions>
+					<execution>
+						<id>copy-resources</id>
+						<!-- here the phase you need -->
+						<phase>package</phase>
+						<goals>
+							<goal>run</goal>
+						</goals>
+						<configuration>
+							<tasks>
+								<copy todir="${basedir}/../../${dist.dir}/connector/mysqlcdc/"
+									  file="${basedir}/target/${project.artifactId}-${project.version}.jar"/>
+								<!--suppress UnresolvedMavenProperty -->
+								<move file="${basedir}/../../${dist.dir}/connector/mysqlcdc/${project.artifactId}-${project.version}.jar"
+									  tofile="${basedir}/../../${dist.dir}/connector/mysqlcdc/${project.artifactId}.jar"/>
+								<delete>
+									<!--suppress UnresolvedMavenProperty -->
+									<fileset dir="${basedir}/../../${dist.dir}/connector/mysqlcdc/"
+											 includes="${project.artifactId}-*.jar"
+											 excludes="${project.artifactId}.jar"/>
+								</delete>
+							</tasks>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
+</project>