[Bitsail][PG-CDC] add postgres cdc connector

Author: xubo.huster

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;
@@ -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-jdbc-base/pom.xml

@@ -1,4 +1,19 @@
 <?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">
@@ -44,7 +59,6 @@
             <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

@@ -1,14 +1,29 @@
+/*
+ * 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.Builder;
 import lombok.Getter;
 import org.apache.commons.lang.StringUtils;
 
@@ -17,7 +32,6 @@
 import java.util.Properties;
 
 @Getter
-@Builder
 public abstract class AbstractJdbcDebeziumConfig {
 
   private static final long serialVersionUID = 1L;
@@ -32,28 +46,26 @@ public abstract class AbstractJdbcDebeziumConfig {
   // debezium configuration
   private final Properties dbzProperties;
   private final Configuration dbzConfiguration;
+  private final RelationalDatabaseConnectorConfig dbzJdbcConnectorConfig;
+  private String dbName;
 
-  public static AbstractJdbcDebeziumConfig fromBitSailConf(BitSailConfiguration jobConf) {
+  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);
-    Properties props = extractProps(jobConf);
-    String username = jobConf.getNecessaryOption(BinlogReaderOptions.USER_NAME, BinlogReaderErrorCode.REQUIRED_VALUE);
-    String password = jobConf.getNecessaryOption(BinlogReaderOptions.PASSWORD, BinlogReaderErrorCode.REQUIRED_VALUE);
+    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(props, connectionInfo, username, password, timezone);
+    fillConnectionInfo(this.dbzProperties, connectionInfo, timezone);
 
-    Configuration config = Configuration.from(props);
-    return AbstractJdbcDebeziumConfig.builder()
-            .hostname(connectionInfo.getHost())
-            .port(connectionInfo.getPort())
-            .username(username)
-            .password(password)
-            .dbzProperties(props)
-            .dbzConfiguration(config)
-            .build();
+    this.dbzConfiguration = Configuration.from(this.dbzProperties);
+    this.dbzJdbcConnectorConfig = getJdbcConnectorConfig(this.dbzConfiguration);
   }
 
   public static Properties extractProps(BitSailConfiguration jobConf) {
@@ -65,14 +77,15 @@ public static Properties extractProps(BitSailConfiguration jobConf) {
     return props;
   }
 
-  public abstract RelationalDatabaseConnectorConfig getConnectorConfig();
+  public abstract RelationalDatabaseConnectorConfig getJdbcConnectorConfig(Configuration config);
 
-  public static void fillConnectionInfo(Properties props, ConnectionInfo connectionInfo, String username, String password, String timezone) {
-    props.put("database.hostname", connectionInfo.getHost());
-    props.put("database.port", String.valueOf(connectionInfo.getPort()));
-    props.put("database.user", username);
-    props.put("database.password", password);
-    props.put("database.serverTimezone", ZoneId.of(timezone).toString());
+  public void fillConnectionInfo(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";
+}

bitsail-connectors/connector-cdc/connector-cdc-jdbc-base/src/main/java/com/bytedance/bitsail/connector/cdc/jdbc/source/reader/AbstractJdbcChangeEventReader.java renamed to bitsail-connectors/connector-cdc/connector-cdc-jdbc-base/src/main/java/com/bytedance/bitsail/connector/cdc/jdbc/source/reader/AbstractJdbcChangeEventSplitReader.java

@@ -1,16 +1,31 @@
+/*
+ * 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.reader;
 
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
 import com.bytedance.bitsail.connector.cdc.jdbc.source.config.AbstractJdbcDebeziumConfig;
-import com.bytedance.bitsail.connector.cdc.jdbc.source.streaming.SplitChangeEventStreamingTaskContext;
+import com.bytedance.bitsail.connector.cdc.jdbc.source.streaming.AbstractSplitChangeEventStreamingTaskContext;
 import com.bytedance.bitsail.connector.cdc.jdbc.source.streaming.SplitChangeEventStreamingTaskController;
 import com.bytedance.bitsail.connector.cdc.source.reader.BinlogSplitReader;
 import com.bytedance.bitsail.connector.cdc.source.split.BinlogSplit;
 import com.bytedance.bitsail.common.row.Row;
 import io.debezium.connector.base.ChangeEventQueue;
-import io.debezium.connector.common.CdcSourceTaskContext;
 import io.debezium.pipeline.DataChangeEvent;
 import io.debezium.relational.RelationalDatabaseConnectorConfig;
+import lombok.Getter;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,58 +38,54 @@
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
-public abstract class AbstractJdbcChangeEventReader implements BinlogSplitReader<Row> {
-  private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcChangeEventReader.class);
+@Getter
+public abstract class AbstractJdbcChangeEventSplitReader implements BinlogSplitReader<Row> {
+  private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcChangeEventSplitReader.class);
 
-  private final AbstractJdbcDebeziumConfig jdbcDebeziumConfig;
+  protected final AbstractJdbcDebeziumConfig jdbcDebeziumConfig;
 
-  private ChangeEventQueue<DataChangeEvent> queue;
+  protected ChangeEventQueue<DataChangeEvent> queue;
 
-  private RelationalDatabaseConnectorConfig connectorConfig;
+  protected RelationalDatabaseConnectorConfig connectorConfig;
 
-  private List<SourceRecord> batch;
+  protected List<SourceRecord> batch;
 
-  private Iterator<SourceRecord> recordIterator;
+  protected Iterator<SourceRecord> recordIterator;
 
-  private CdcSourceTaskContext taskContext;
+  protected Map<String, ?> offset;
 
-  private Map<String, ?> offset;
+  protected SplitChangeEventStreamingTaskController splitChangeEventStreamingTaskController;
 
-  private SplitChangeEventStreamingTaskController splitChangeEventStreamingTaskController;
-
-  private SplitChangeEventStreamingTaskContext splitChangeEventStreamingTaskContext;
+  protected AbstractSplitChangeEventStreamingTaskContext splitChangeEventStreamingTaskContext;
 
   private final int subtaskId;
 
-  public AbstractJdbcChangeEventReader(BitSailConfiguration jobConf, int subtaskId) {
+  public AbstractJdbcChangeEventSplitReader(BitSailConfiguration jobConf, int subtaskId) {
     jdbcDebeziumConfig = getJdbcDebeziumConfig(jobConf);
-    connectorConfig = jdbcDebeziumConfig.getConnectorConfig();
+    connectorConfig = jdbcDebeziumConfig.getDbzJdbcConnectorConfig();
     this.subtaskId = subtaskId;
     this.offset = new HashMap<>();
   }
 
-  public AbstractJdbcDebeziumConfig getJdbcDebeziumConfig(BitSailConfiguration jobConf) {
-    return AbstractJdbcDebeziumConfig.fromBitSailConf(jobConf);
-  }
-
-  public abstract SplitChangeEventStreamingTaskContext getSplitReaderTaskContext();
+  public abstract AbstractJdbcDebeziumConfig getJdbcDebeziumConfig(BitSailConfiguration jobConf);
 
-  public abstract void testConnectionAndValidBinlogConfiguration(RelationalDatabaseConnectorConfig connectorConfig) throws IOException;
+  public abstract AbstractSplitChangeEventStreamingTaskContext getSplitReaderTaskContext(BinlogSplit split, RelationalDatabaseConnectorConfig connectorConfig);
 
   public void inititialzeSplitReader(BinlogSplit split) {
-    splitChangeEventStreamingTaskContext = getSplitReaderTaskContext();
+    splitChangeEventStreamingTaskContext = getSplitReaderTaskContext(split, connectorConfig);
     this.offset = new HashMap<>();
     this.queue = new ChangeEventQueue.Builder<DataChangeEvent>()
             .pollInterval(connectorConfig.getPollInterval())
             .maxBatchSize(connectorConfig.getMaxBatchSize())
             .maxQueueSize(connectorConfig.getMaxQueueSize())
             .maxQueueSizeInBytes(connectorConfig.getMaxQueueSizeInBytes())
-            .loggingContextSupplier(() -> taskContext.configureLoggingContext(splitChangeEventStreamingTaskContext.threadNamePrefix()))
+            .loggingContextSupplier(() -> splitChangeEventStreamingTaskContext.getDbzTaskContext()
+                    .configureLoggingContext(splitChangeEventStreamingTaskContext.threadNamePrefix()))
             .buffering()
             .build();
     this.batch = new ArrayList<>();
     this.recordIterator = this.batch.iterator();
-    splitChangeEventStreamingTaskContext.initializeSplitReaderTaskContext(connectorConfig, this.queue);
+    splitChangeEventStreamingTaskContext.attachStreamingToQueue(this.queue);
     splitChangeEventStreamingTaskController = new SplitChangeEventStreamingTaskController(splitChangeEventStreamingTaskContext, this.subtaskId);
   }
 
@@ -136,7 +147,7 @@ public Row poll() {
    * @throws Exception
    */
   @Override
-  public boolean hasNext() throws Exception {
+  public boolean hasNext() {
     if (this.recordIterator.hasNext()) {
       return true;
     } else {
@@ -149,20 +160,24 @@ public boolean isCompleted() {
     return !splitChangeEventStreamingTaskController.isRunning();
   }
 
-  private boolean pollNextBatch() throws InterruptedException {
+  private boolean pollNextBatch() {
     if (splitChangeEventStreamingTaskController.isRunning()) {
-      List<DataChangeEvent> dbzRecords = queue.poll();
-      while (dbzRecords.isEmpty()) {
-        //sleep 10s
-        LOG.info("No record found, sleep for 5s in reader");
-        TimeUnit.SECONDS.sleep(5);
-        dbzRecords = queue.poll();
-      }
-      this.batch = new ArrayList<>();
-      for (DataChangeEvent event : dbzRecords) {
-        this.batch.add(event.getRecord());
+      try {
+        List<DataChangeEvent> dbzRecords = queue.poll();
+        while (dbzRecords.isEmpty()) {
+          //sleep 10s
+          LOG.info("No record found, sleep for 5s in reader");
+          TimeUnit.SECONDS.sleep(5);
+          dbzRecords = queue.poll();
+        }
+        this.batch = new ArrayList<>();
+        for (DataChangeEvent event : dbzRecords) {
+          this.batch.add(event.getRecord());
+        }
+        this.recordIterator = this.batch.iterator();
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
       }
-      this.recordIterator = this.batch.iterator();
       return true;
     }
     return false;