Merge pull request #756 from Altinity/745-comparison-of-single-threaded-vs-queue-approach-for-memory-usage-snapshot

Added functionality to replicate in single threaded mode based on configuration without using a Queue

Author: subkanthi

README.md

@@ -59,6 +59,7 @@ First two are good tutorials on MySQL and PostgreSQL respectively.
 * [Logging](doc/logging.md)
 * [Production Setup](doc/production_setup.md)
 * [Adding new tables(Incremental Snapshot)](doc/incremental_snapshot.md)
+* [Configuration](doc/configuration.md)
 
 ### Operations
 

doc/configuration.md

@@ -65,6 +65,11 @@ The maximum number of rows that the connector fetches and reads into memory when
 
 **snapshot.max.threads**: Increase this number from 1 to a higher value to enable parallel snapshotting.
 
+**Single Threaded (Low Memory/Slow replication)**:
+By setting the `single.threaded: true` configuration variable in `config.yml`, the replication will skip the sink connector queue and threadpool
+and will insert batches directly from the debezium queue.
+This mode will work on lower memory setup but will increase the replication speed.
+
 ## PostgreSQL Production Setup
 
 One of the common problems with PostgreSQL is the WAL size increasing.

doc/production_setup.md

@@ -13,6 +13,7 @@
 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.model.ClickHouseStruct;
 import com.altinity.clickhouse.sink.connector.model.DBCredentials;
 import com.clickhouse.jdbc.ClickHouseConnection;
@@ -29,6 +30,7 @@
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.checkerframework.checker.units.qual.C;
 import org.json.simple.JSONArray;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.ParseException;
@@ -81,6 +83,8 @@ public class DebeziumChangeEventCapture {
     // Keep one clickhouse connection.
     private ClickHouseConnection conn;
 
+    ClickHouseBatchWriter singleThreadedWriter;
+
     public DebeziumChangeEventCapture() {
         singleThreadDebeziumEventExecutor = Executors.newFixedThreadPool(1);
     }
@@ -596,7 +600,7 @@ public void handleBatch(List<ChangeEvent<SourceRecord, SourceRecord>> list,
 
 
                     if(batch.size() > 0) {
-                        appendToRecords(batch);
+                        appendToRecords(batch, config);
                     }
                 }
             });
@@ -775,23 +779,33 @@ DBCredentials parseDBConfiguration(ClickHouseSinkConnectorConfig config) {
      */
     private void setupProcessingThread(ClickHouseSinkConnectorConfig config) {
 
-        // Setup separate thread to read messages from shared buffer.
-        // this.records = new ConcurrentLinkedQueue<>();
-        //this.runnable = new ClickHouseBatchRunnable(this.records, config, new HashMap());
-        ThreadFactory namedThreadFactory =
-                new ThreadFactoryBuilder().setNameFormat("Sink Connector thread-pool-%d").build();
-        this.executor = new ClickHouseBatchExecutor(config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()), namedThreadFactory);
-        for(int i = 0; i < config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()); i++) {
-            this.executor.scheduleAtFixedRate(new ClickHouseBatchRunnable(this.records, config, new HashMap()), 0,
-                    config.getLong(ClickHouseSinkConnectorConfigVariables.BUFFER_FLUSH_TIME.toString()), TimeUnit.MILLISECONDS);
+        if(config.getBoolean(ClickHouseSinkConnectorConfigVariables.SINGLE_THREADED.toString())) {
+            log.info("********* Running in Single Threaded mode *********");
+            singleThreadedWriter = new ClickHouseBatchWriter(config, new HashMap());
         }
+
+            ThreadFactory namedThreadFactory =
+                    new ThreadFactoryBuilder().setNameFormat("Sink Connector thread-pool-%d").build();
+            this.executor = new ClickHouseBatchExecutor(config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()), namedThreadFactory);
+            for (int i = 0; i < config.getInt(ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString()); i++) {
+                this.executor.scheduleAtFixedRate(new ClickHouseBatchRunnable(this.records, config, new HashMap()), 0,
+                        config.getLong(ClickHouseSinkConnectorConfigVariables.BUFFER_FLUSH_TIME.toString()), TimeUnit.MILLISECONDS);
+            }
+
         //this.executor.scheduleAtFixedRate(this.runnable, 0, config.getLong(ClickHouseSinkConnectorConfigVariables.BUFFER_FLUSH_TIME.toString()), TimeUnit.MILLISECONDS);
     }
 
-    private void appendToRecords(List<ClickHouseStruct> convertedRecords) {
+    private void appendToRecords(List<ClickHouseStruct> convertedRecords, ClickHouseSinkConnectorConfig config) {
 
-        synchronized (this.records) {
-            this.records.add(convertedRecords);
+        // If config is set to single threaded.
+        if(config.getBoolean(ClickHouseSinkConnectorConfigVariables.SINGLE_THREADED.toString())) {
+            singleThreadedWriter.persistRecords(convertedRecords);
+
+        } else {
+
+            synchronized (this.records) {
+                this.records.add(convertedRecords);
+            }
         }
 
 

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

@@ -450,6 +450,16 @@ static ConfigDef newConfigDef() {
                         6,
                         ConfigDef.Width.NONE,
                         ClickHouseSinkConnectorConfigVariables.MAX_QUEUE_SIZE.toString())
+                .define(
+                        ClickHouseSinkConnectorConfigVariables.SINGLE_THREADED.toString(),
+                        Type.BOOLEAN,
+                        false,
+                        Importance.HIGH,
+                        "Single threaded mode",
+                        CONFIG_GROUP_CONNECTOR_CONFIG,
+                        6,
+                        ConfigDef.Width.NONE,
+                        ClickHouseSinkConnectorConfigVariables.SINGLE_THREADED.toString())
                 .define(
                         ClickHouseSinkConnectorConfigVariables.REPLICA_STATUS_VIEW.toString(),
                         Type.STRING,

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java

@@ -73,7 +73,9 @@ public enum ClickHouseSinkConnectorConfigVariables {
     JDBC_PARAMETERS("clickhouse.jdbc.params"),
 
     REPLICA_STATUS_VIEW("replica.status.view"),
-    MAX_QUEUE_SIZE("sink.connector.max.queue.size");
+    MAX_QUEUE_SIZE("sink.connector.max.queue.size"),
+
+    SINGLE_THREADED("single.threaded");
 
     private String label;