Use Hikari Connection Pool for handling Clickhouse connections

sink-connector-lightweight/dependency-reduced-pom.xml

@@ -162,10 +162,6 @@
           <artifactId>oracle-xe</artifactId>
           <groupId>org.testcontainers</groupId>
         </exclusion>
-        <exclusion>
-          <artifactId>junit-platform-launcher</artifactId>
-          <groupId>org.junit.platform</groupId>
-        </exclusion>
         <exclusion>
           <artifactId>junit-jupiter</artifactId>
           <groupId>org.junit.jupiter</groupId>
@@ -207,7 +203,7 @@
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>testcontainers</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
       <exclusions>
         <exclusion>
@@ -231,7 +227,7 @@
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>jdbc</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
       <exclusions>
         <exclusion>
@@ -243,37 +239,37 @@
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>junit-jupiter</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>clickhouse</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>mysql</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>postgresql</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>mariadb</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>mongodb</artifactId>
-      <version>1.19.1</version>
+      <version>1.20.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -289,10 +285,6 @@
       <version>5.8.1</version>
       <scope>test</scope>
       <exclusions>
-        <exclusion>
-          <artifactId>junit-platform-engine</artifactId>
-          <groupId>org.junit.platform</groupId>
-        </exclusion>
         <exclusion>
           <artifactId>junit-jupiter-api</artifactId>
           <groupId>org.junit.jupiter</groupId>
@@ -317,10 +309,38 @@
           <artifactId>junit-jupiter-api</artifactId>
           <groupId>org.junit.jupiter</groupId>
         </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.platform</groupId>
+      <artifactId>junit-platform-launcher</artifactId>
+      <version>1.8.2</version>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <artifactId>apiguardian-api</artifactId>
+          <groupId>org.apiguardian</groupId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.platform</groupId>
+      <artifactId>junit-platform-engine</artifactId>
+      <version>1.8.2</version>
+      <scope>test</scope>
+      <exclusions>
         <exclusion>
-          <artifactId>junit-platform-launcher</artifactId>
+          <artifactId>opentest4j</artifactId>
+          <groupId>org.opentest4j</groupId>
+        </exclusion>
+        <exclusion>
+          <artifactId>junit-platform-commons</artifactId>
           <groupId>org.junit.platform</groupId>
         </exclusion>
+        <exclusion>
+          <artifactId>apiguardian-api</artifactId>
+          <groupId>org.apiguardian</groupId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>
@@ -356,7 +376,7 @@
   <properties>
     <quarkus.platform.version>2.14.0.Final</quarkus.platform.version>
     <quarkus.platform.artifact-id>quarkus-bom</quarkus.platform.artifact-id>
-    <version.testcontainers>1.19.1</version.testcontainers>
+    <version.testcontainers>1.20.4</version.testcontainers>
     <surefire-plugin.version>3.0.0-M7</surefire-plugin.version>
     <version.kafka>3.8.0</version.kafka>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

sink-connector-lightweight/docker/config.yml

@@ -29,7 +29,7 @@ database.server.name: "ER54"
 
 # database.include.list An optional list of regular expressions that match database names to be monitored;
 # any database name not included in the whitelist will be excluded from monitoring. By default all databases will be monitored.
-database.include.list: test
+database.include.list: sbtest
 
 # table.include.list An optional list of regular expressions that match fully-qualified table identifiers for tables to be monitored;
 table.include.list: ""
@@ -179,6 +179,8 @@ ignore.ddl.regex: "(?i)(ANALYZE PARTITION).*"
 #Metrics (Prometheus target), required for Grafana Dashboard
 metrics.enable: "true"
 
+connection.pool.max.size: 1000
+
 # Skip schema history capturing, use the following configuration
 # to reduce slow startup when replicating dbs with large number of tables
 #schema.history.internal.store.only.captured.tables.ddl: "true"
@@ -189,4 +191,4 @@ use.nongraceful.disconnect: "true"
 database.keep.alive.interval.`ms: "30000"  #Send keepalive every 30 seconds
 database.connection.reconnect.backoff.ms: "1000"
 database.connection.reconnect.backoff.max.ms: "10000"
-database.ssl.mode: "disabled"
+database.ssl.mode: "disabled"

sink-connector-lightweight/pom.xml

@@ -15,7 +15,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <version.debezium>3.0.0.Final</version.debezium>
         <version.junit>5.9.1</version.junit>
-        <version.testcontainers>1.19.1</version.testcontainers>
+        <version.testcontainers>1.20.4</version.testcontainers>
         <version.checkstyle.plugin>3.1.1</version.checkstyle.plugin>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
@@ -233,6 +233,11 @@
             <!-- <scope>test</scope>-->
         </dependency>
 
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP</artifactId>
+            <version>6.0.0</version>
+        </dependency>
         <dependency>
             <groupId>org.antlr</groupId>
             <artifactId>antlr4-runtime</artifactId>
@@ -334,6 +339,18 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.junit.platform</groupId>
+            <artifactId>junit-platform-launcher</artifactId>
+            <version>1.8.2</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.platform</groupId>
+            <artifactId>junit-platform-engine</artifactId>
+            <version>1.8.2</version>
+            <scope>test</scope>
+        </dependency>
         <dependency> <!-- necessary for Java 9+ -->
             <groupId>org.apache.tomcat</groupId>
             <artifactId>annotations-api</artifactId>

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java

@@ -2,22 +2,21 @@
 
 import com.altinity.clickhouse.debezium.embedded.common.PropertiesHelper;
 import com.altinity.clickhouse.debezium.embedded.config.SinkConnectorLightWeightConfig;
-import com.altinity.clickhouse.debezium.embedded.ddl.parser.DDLParserService;
 import com.altinity.clickhouse.debezium.embedded.ddl.parser.MySQLDDLParserService;
 import com.altinity.clickhouse.debezium.embedded.parser.DebeziumRecordParserService;
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig;
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
 import com.altinity.clickhouse.sink.connector.common.Metrics;
 import com.altinity.clickhouse.sink.connector.db.BaseDbWriter;
 import com.altinity.clickhouse.sink.connector.db.DBMetadata;
+import com.altinity.clickhouse.sink.connector.db.HikariDbSource;
 import com.altinity.clickhouse.sink.connector.db.operations.ClickHouseAlterTable;
 import com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchExecutor;
 import com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchRunnable;
 import com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter;
 import com.altinity.clickhouse.sink.connector.executor.DebeziumOffsetManagement;
 import com.altinity.clickhouse.sink.connector.model.ClickHouseStruct;
 import com.altinity.clickhouse.sink.connector.model.DBCredentials;
-import com.clickhouse.jdbc.ClickHouseConnection;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.debezium.config.CommonConnectorConfig;
@@ -41,6 +40,7 @@
 import org.json.simple.parser.ParseException;
 
 import java.io.IOException;
+import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
@@ -89,7 +89,7 @@ public class DebeziumChangeEventCapture {
     DebeziumEngine<ChangeEvent<SourceRecord, SourceRecord>> engine;
 
     // Keep one clickhouse connection.
-    private ClickHouseConnection conn;
+    private Connection conn;
 
     ClickHouseBatchWriter singleThreadedWriter;
 
@@ -197,7 +197,8 @@ private String getDatabaseName(SourceRecord sr) {
     private BaseDbWriter createWriter(ClickHouseSinkConnectorConfig config, String databaseName) {
         DBCredentials dbCredentials = parseDBConfiguration(config);
         String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(), databaseName);
-        ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1", dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+        Connection conn = BaseDbWriter.createConnection(jdbcUrl, BaseDbWriter.DATABASE_CLIENT_NAME, dbCredentials.getUserName(), dbCredentials.getPassword(),
+                databaseName, config);
         return new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(), databaseName, dbCredentials.getUserName(), dbCredentials.getPassword(), config, conn);
     }
 
@@ -270,11 +271,17 @@ private ClickHouseStruct processEveryChangeRecord(Properties props, ChangeEvent<
                 if (DDL != null && DDL.isEmpty() == false)
                 {
                     log.info("***** DDL received, Flush all existing records");
-                    this.executor.shutdown();
-                    this.executor.awaitTermination(60, TimeUnit.SECONDS);
+//                    this.executor.shutdown();
+//                    this.executor.awaitTermination(60, TimeUnit.SECONDS);
+//                    //HikariDbSource.closeDatabaseConnection(BaseDbWriter.SYSTEM_DB);
+                    this.executor.pause();
 
+                    //this.writer = null;
+                    //this.writer = null;
                     performDDLOperation(DDL, props, sr, config, recordCommitter, record, lastRecordInBatch);
-                    setupProcessingThread(config);
+                    this.executor.resume();
+                    //HikariDbSource.closeDatabaseConnection(BaseDbWriter.SYSTEM_DB);
+                    //setupProcessingThread(config);
                 }
 
             } else {
@@ -383,18 +390,19 @@ private void createDatabaseForDebeziumStorage(ClickHouseSinkConnectorConfig conf
                 DBCredentials dbCredentials = parseDBConfiguration(config);
 
                 String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
-                        "system");
-                ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1", dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+                        BaseDbWriter.SYSTEM_DB);
+                Connection conn = BaseDbWriter.createConnection(jdbcUrl, BaseDbWriter.DATABASE_CLIENT_NAME,
+                        dbCredentials.getUserName(), dbCredentials.getPassword(), BaseDbWriter.SYSTEM_DB, config);
                 BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
-                        "system", dbCredentials.getUserName(),
+                        BaseDbWriter.SYSTEM_DB, dbCredentials.getUserName(),
                         dbCredentials.getPassword(), config, conn);
 
                 Pair<String, String> tableNameDatabaseName = getDebeziumOffsetStorageDatabaseName(props);
                 String databaseName = tableNameDatabaseName.getRight();
 
                 String createDbQuery = String.format("create database if not exists %s", databaseName);
                 log.info("CREATING DEBEZIUM STORAGE Database: " + createDbQuery);
-                writer.executeQuery(createDbQuery);
+                writer.executeSystemQuery(createDbQuery);
 
                 break;
             } catch (Exception e) {
@@ -422,13 +430,14 @@ private void createSchemaHistoryTable(ClickHouseSinkConnectorConfig config, Prop
         DBCredentials dbCredentials = parseDBConfiguration(config);
         String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
                 "system");
-        ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+        Connection conn = BaseDbWriter.createConnection(jdbcUrl, BaseDbWriter.DATABASE_CLIENT_NAME,
+                dbCredentials.getUserName(), dbCredentials.getPassword(), BaseDbWriter.SYSTEM_DB, config);
         BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                 "system", dbCredentials.getUserName(),
                 dbCredentials.getPassword(), config, conn);
 
         try {
-            writer.executeQuery(createSchemaHistoryTable);
+            writer.executeSystemQuery(createSchemaHistoryTable);
         } catch(Exception e) {
             log.error("Error creating schema history table", e);
         }
@@ -449,7 +458,8 @@ private void createViewForShowReplicaStatus(ClickHouseSinkConnectorConfig config
 
         String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
                 "system");
-        ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+        Connection conn = BaseDbWriter.createConnection(jdbcUrl, BaseDbWriter.DATABASE_CLIENT_NAME,
+                dbCredentials.getUserName(), dbCredentials.getPassword(), BaseDbWriter.SYSTEM_DB, config);
         BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                 "system", dbCredentials.getUserName(),
                 dbCredentials.getPassword(), config, conn);
@@ -462,7 +472,7 @@ private void createViewForShowReplicaStatus(ClickHouseSinkConnectorConfig config
         // Remove quotes.
         formattedView = formattedView.replace("\"", "");
         try {
-            writer.executeQuery(formattedView);
+            writer.executeSystemQuery(formattedView);
         } catch(Exception e) {
             log.error("**** Error creating VIEW **** " + formattedView);
         }
@@ -539,8 +549,9 @@ public String getDebeziumStorageStatus(ClickHouseSinkConnectorConfig config, Pro
             log.error("**** Connection to ClickHouse is not established, re-initiating ****");
             String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
                     databaseName);
-            ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",
-                    dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+            Connection conn = BaseDbWriter.createConnection(jdbcUrl, BaseDbWriter.DATABASE_CLIENT_NAME,
+                    dbCredentials.getUserName(), dbCredentials.getPassword(),
+                    BaseDbWriter.SYSTEM_DB, config);
             writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                     databaseName, dbCredentials.getUserName(),
                     dbCredentials.getPassword(), config, conn);
@@ -956,6 +967,7 @@ DBCredentials parseDBConfiguration(ClickHouseSinkConnectorConfig config) {
      */
     private void setupProcessingThread(ClickHouseSinkConnectorConfig config) {
 
+
         if(config.getBoolean(ClickHouseSinkConnectorConfigVariables.SINGLE_THREADED.toString())) {
             log.info("********* Running in Single Threaded mode *********");
             singleThreadedWriter = new ClickHouseBatchWriter(config, new HashMap());

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumOffsetStorage.java

@@ -51,7 +51,7 @@ public void deleteOffsetStorageRow(String offsetKey,
 
         // String connectorName = config.getString("connector.name");
         String debeziumStorageStatusQuery = String.format("delete from %s where offset_key='%s'" , tableName, offsetKey);
-        writer.executeQuery(debeziumStorageStatusQuery);
+        writer.executeSystemQuery(debeziumStorageStatusQuery);
     }
 
     /**
@@ -67,7 +67,7 @@ public void deleteSchemaHistoryTable(String offsetKey,
 
         String debeziumStorageStatusQuery = String.format("delete from `%s` where JSONExtractRaw(JSONExtractRaw(history_data,'source'), 'server')='%s'" , tableName, offsetKey);
         log.info("Deleting schema history table query: " + debeziumStorageStatusQuery);
-        writer.executeQuery(debeziumStorageStatusQuery);
+        writer.executeSystemQuery(debeziumStorageStatusQuery);
     }
     /**
      * Function to get the latest timestamp of the record in the table
@@ -81,7 +81,7 @@ public String getDebeziumLatestRecordTimestamp(Properties props, BaseDbWriter wr
                 JdbcOffsetBackingStoreConfig.PROP_TABLE_NAME.name());
 
         String debeziumLatestRecordTimestampQuery = String.format("select max(record_insert_ts) from %s" , tableName);
-        return writer.executeQuery(debeziumLatestRecordTimestampQuery);
+        return writer.executeSystemQuery(debeziumLatestRecordTimestampQuery);
     }
 
     public String getDebeziumStorageStatusQuery(
@@ -92,7 +92,7 @@ public String getDebeziumStorageStatusQuery(
         String offsetKey = getOffsetKey(props);
         // String connectorName = config.getString("connector.name");
         String debeziumStorageStatusQuery = String.format("select offset_val from %s where offset_key='%s'" , tableName, offsetKey);
-        return writer.executeQuery(debeziumStorageStatusQuery);
+        return writer.executeSystemQuery(debeziumStorageStatusQuery);
     }
 
     /**

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/Constants.java

@@ -50,6 +50,7 @@ public class Constants {
 
 
     public static final String CREATE_DATABASE = "CREATE DATABASE IF NOT EXISTS %s";
+    public static final String DROP_DATABASE = "DROP DATABASE IF EXISTS %s";
 
     public static final String DROP_COLUMN = "DROP COLUMN %s";