[Bitsail][Connector] support pg cdc

bitsail-base/src/main/java/com/bytedance/bitsail/base/connector/reader/v1/SourceReader.java

@@ -23,7 +23,7 @@
 
 public interface SourceReader<T, SplitT extends SourceSplit> extends Serializable, AutoCloseable {
 
-  void start();
+  void start() throws Exception;
 
   void pollNext(SourcePipeline<T> pipeline) throws Exception;
 

bitsail-common/src/main/java/com/bytedance/bitsail/common/option/ReaderOptions.java

@@ -88,6 +88,10 @@ interface BaseReaderOptions {
         key(READER_PREFIX + "db_name")
             .noDefaultValue(String.class);
 
+    ConfigOption<String> CONNECTION_TIMEZONE =
+            key(READER_PREFIX + "connection_timezone")
+                    .defaultValue("UTC");
+
     ConfigOption<String> TABLE_NAME =
         key(READER_PREFIX + "table_name")
             .noDefaultValue(String.class);

bitsail-connectors/connector-cdc/connector-cdc-base/pom.xml

@@ -33,7 +33,6 @@
     </properties>
 
     <dependencies>
-
     </dependencies>
 
 </project>

bitsail-connectors/connector-cdc/connector-cdc-base/src/main/java/com/bytedance/bitsail/connector/cdc/source/reader/BinlogSourceReader.java

@@ -27,6 +27,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.ArrayDeque;
 import java.util.List;
 import java.util.Queue;
@@ -56,7 +57,7 @@ public BinlogSourceReader(BitSailConfiguration jobConf, SourceReader.Context rea
   public abstract BinlogSplitReader<Row> getReader();
 
   @Override
-  public void start() {
+  public void start() throws Exception{
     //start debezium streaming reader and send data to queue
   }
 
@@ -115,7 +116,7 @@ public void close() {
     }
   }
 
-  private void submitSplit() {
+  private void submitSplit() throws IOException, InterruptedException {
     if (!remainSplits.isEmpty()) {
       BinlogSplit curSplit = remainSplits.poll();
       LOG.info("submit split to binlog reader: {}, size of the remaining splits: {}", curSplit.toString(), remainSplits.size());

bitsail-connectors/connector-cdc/connector-cdc-base/src/main/java/com/bytedance/bitsail/connector/cdc/source/reader/BinlogSplitReader.java

@@ -18,11 +18,12 @@
 
 import com.bytedance.bitsail.connector.cdc.source.split.BinlogSplit;
 
+import java.io.IOException;
 import java.io.Serializable;
 import java.util.Map;
 
 public interface BinlogSplitReader<T> extends Serializable {
-  void readSplit(BinlogSplit split);
+  void readSplit(BinlogSplit split) throws IOException, InterruptedException;
 
   Map<String, String> getOffset();
 

bitsail-connectors/connector-cdc/connector-cdc-jdbc-base/pom.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+  ~
+  ~ Licensed 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>
+        <groupId>com.bytedance.bitsail</groupId>
+        <artifactId>connector-cdc</artifactId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>connector-cdc-jdbc-base</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <debezium.version>1.6.4.Final</debezium.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>com.bytedance.bitsail</groupId>
+            <artifactId>connector-cdc-base</artifactId>
+            <version>${revision}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-embedded</artifactId>
+            <version>${debezium.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>jakarta.activation</groupId>
+                    <artifactId>jakarta.activation-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.kafka</groupId>
+                    <artifactId>kafka-clients</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-core</artifactId>
+            <version>${debezium.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

bitsail-connectors/connector-cdc/connector-cdc-jdbc-base/src/main/java/com/bytedance/bitsail/connector/cdc/jdbc/source/config/AbstractJdbcDebeziumConfig.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+ *
+ * Licensed 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.bytedance.bitsail.connector.cdc.jdbc.source.config;
+
+import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
+import com.bytedance.bitsail.connector.cdc.error.BinlogReaderErrorCode;
+import com.bytedance.bitsail.connector.cdc.jdbc.source.constant.DebeziumConstant;
+import com.bytedance.bitsail.connector.cdc.model.ClusterInfo;
+import com.bytedance.bitsail.connector.cdc.model.ConnectionInfo;
+import com.bytedance.bitsail.connector.cdc.option.BinlogReaderOptions;
+
+import io.debezium.config.Configuration;
+import io.debezium.relational.RelationalDatabaseConnectorConfig;
+import lombok.Getter;
+import org.apache.commons.lang.StringUtils;
+
+import java.time.ZoneId;
+import java.util.List;
+import java.util.Properties;
+
+@Getter
+public abstract class AbstractJdbcDebeziumConfig {
+
+  private static final long serialVersionUID = 1L;
+
+  public static final String DEBEZIUM_PREFIX = "job.reader.debezium.";
+
+  private final String hostname;
+  private final int port;
+  private final String username;
+  private final String password;
+
+  // debezium configuration
+  private final Properties dbzProperties;
+  private final Configuration dbzConfiguration;
+  private final RelationalDatabaseConnectorConfig dbzJdbcConnectorConfig;
+  private String dbName;
+
+  public AbstractJdbcDebeziumConfig(BitSailConfiguration jobConf) {
+    List<ClusterInfo> clusterInfo = jobConf.getNecessaryOption(BinlogReaderOptions.CONNECTIONS, BinlogReaderErrorCode.REQUIRED_VALUE);
+    //Only support one DB
+    assert (clusterInfo.size() == 1);
+    ConnectionInfo connectionInfo = clusterInfo.get(0).getMaster();
+    assert (connectionInfo != null);
+    this.dbzProperties = extractProps(jobConf);
+    this.hostname = connectionInfo.getHost();
+    this.port = connectionInfo.getPort();
+    this.username = jobConf.getNecessaryOption(BinlogReaderOptions.USER_NAME, BinlogReaderErrorCode.REQUIRED_VALUE);
+    this.password = jobConf.getNecessaryOption(BinlogReaderOptions.PASSWORD, BinlogReaderErrorCode.REQUIRED_VALUE);
+    this.dbName = jobConf.getNecessaryOption(BinlogReaderOptions.DB_NAME, BinlogReaderErrorCode.REQUIRED_VALUE);
+    String timezone = jobConf.get(BinlogReaderOptions.CONNECTION_TIMEZONE);
+    fillConnectionInfo(jobConf, this.dbzProperties, connectionInfo, timezone);
+
+    this.dbzConfiguration = Configuration.from(this.dbzProperties);
+    this.dbzJdbcConnectorConfig = getJdbcConnectorConfig(this.dbzConfiguration);
+  }
+
+  public static Properties extractProps(BitSailConfiguration jobConf) {
+    Properties props = new Properties();
+    jobConf.getKeys().stream()
+            .filter(s -> s.startsWith(DEBEZIUM_PREFIX))
+            .map(s -> StringUtils.substringAfter(s, DEBEZIUM_PREFIX))
+            .forEach(s -> props.setProperty(s, jobConf.getString(DEBEZIUM_PREFIX + s)));
+    return props;
+  }
+
+  public abstract RelationalDatabaseConnectorConfig getJdbcConnectorConfig(Configuration config);
+
+  public void fillConnectionInfo(BitSailConfiguration jobConf, Properties props, ConnectionInfo connectionInfo, String timezone) {
+    props.put(DebeziumConstant.DATABASE_HOSTNAME, connectionInfo.getHost());
+    props.put(DebeziumConstant.DATABASE_PORT, String.valueOf(connectionInfo.getPort()));
+    props.put(DebeziumConstant.DATABASE_USER, username);
+    props.put(DebeziumConstant.DATABASE_PASSWORD, password);
+    props.put(DebeziumConstant.DATABASE_NAME, dbName);
+    props.put(DebeziumConstant.DATABASE_TIMEZONE, ZoneId.of(timezone).toString());
+  }
+
+}

bitsail-connectors/connector-cdc/connector-cdc-jdbc-base/src/main/java/com/bytedance/bitsail/connector/cdc/jdbc/source/constant/DebeziumConstant.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+ *
+ * Licensed 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.bytedance.bitsail.connector.cdc.jdbc.source.constant;
+
+public class DebeziumConstant {
+  public static final String DATABASE_HOSTNAME = "database.hostname";
+  public static final String DATABASE_PORT = "database.port";
+  public static final String DATABASE_USER = "database.user";
+  public static final String DATABASE_PASSWORD = "database.password";
+  public static final String DATABASE_NAME = "database.dbname";
+  public static final String DATABASE_TIMEZONE = "database.serverTimezone";
+}