Changed from ConcurrentLinkedQueue to LinkedBlockedQueue with the siz…

…e configurable

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

@@ -55,7 +55,7 @@ public class DebeziumChangeEventCapture {
     private ClickHouseBatchRunnable runnable;
 
     // Records grouped by Topic Name
-    private ConcurrentLinkedQueue<List<ClickHouseStruct>> records = new ConcurrentLinkedQueue<>();
+    private LinkedBlockingQueue<List<ClickHouseStruct>> records;
 
 
     private BaseDbWriter writer = null;
@@ -586,6 +586,13 @@ public void connectorStopped() {
     public void setup(Properties props, DebeziumRecordParserService debeziumRecordParserService,
                       DDLParserService ddlParserService, boolean forceStart) throws IOException, ClassNotFoundException {
 
+        // Check if max queue size was defined by the user.
+        if(props.getProperty(ClickHouseSinkConnectorConfigVariables.MAX_QUEUE_SIZE.toString()) != null) {
+            int maxQueueSize = Integer.parseInt(props.getProperty(ClickHouseSinkConnectorConfigVariables.MAX_QUEUE_SIZE.toString()));
+            this.records = new LinkedBlockingQueue<>(maxQueueSize);
+        } else {
+            this.records = new LinkedBlockingQueue<>();
+        }
 
         ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(PropertiesHelper.toMap(props));
         Metrics.initialize(props.getProperty(ClickHouseSinkConnectorConfigVariables.ENABLE_METRICS.toString()),

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

@@ -423,8 +423,19 @@ static ConfigDef newConfigDef() {
                         6,
                         ConfigDef.Width.NONE,
                         ClickHouseSinkConnectorConfigVariables.JDBC_PARAMETERS.toString())
+                // Define the max queue size.
+                .define(
+                        ClickHouseSinkConnectorConfigVariables.MAX_QUEUE_SIZE.toString(),
+                        Type.INT,
+                        500000,
+                        ConfigDef.Range.atLeast(1),
+                        Importance.HIGH,
+                        "The maximum size of the queue",
+                        CONFIG_GROUP_CONNECTOR_CONFIG,
+                        6,
+                        ConfigDef.Width.NONE,
+                        ClickHouseSinkConnectorConfigVariables.MAX_QUEUE_SIZE.toString())
 
-                // ToDo: Add JVM Proxy
                 ;
     }
 }

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

@@ -67,7 +67,9 @@ public enum ClickHouseSinkConnectorConfigVariables {
     RESTART_EVENT_LOOP("restart.event.loop"),
 
     RESTART_EVENT_LOOP_TIMEOUT_PERIOD("restart.event.loop.timeout.period.secs"),
-    JDBC_PARAMETERS("clickhouse.jdbc.params");
+    JDBC_PARAMETERS("clickhouse.jdbc.params"),
+
+    MAX_QUEUE_SIZE("max.queue.size");
 
     private String label;
 

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

@@ -17,10 +17,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 /**
  * <p>Creates sink service instance, takes records loaded from those
@@ -39,7 +36,7 @@ public ClickHouseSinkTask() {
     private ClickHouseBatchExecutor executor;
 
     // Records grouped by Topic Name
-    private ConcurrentLinkedQueue<List<ClickHouseStruct>> records;
+    private LinkedBlockingQueue<List<ClickHouseStruct>> records;
 
     private DeDuplicator deduplicator;
 
@@ -66,7 +63,10 @@ public void start(Map<String, String> config) {
 
         this.id = "task-" + this.config.getLong(ClickHouseSinkConnectorConfigVariables.TASK_ID.toString());
 
-        this.records = new ConcurrentLinkedQueue();
+        // check if the config is defined for MAX_QUEUE_SIZE
+        int maxQueueSize = this.config.getInt(ClickHouseSinkConnectorConfigVariables.MAX_QUEUE_SIZE.toString());
+
+        this.records = new LinkedBlockingQueue<>(maxQueueSize);
         ClickHouseBatchRunnable runnable = new ClickHouseBatchRunnable(this.records, this.config, topic2TableMap);
         ThreadFactory namedThreadFactory =
                 new ThreadFactoryBuilder().setNameFormat("Sink Connector thread-pool-%d").build();

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchRunnable.java

@@ -27,6 +27,7 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * Runnable object that will be called on
@@ -35,7 +36,7 @@
  */
 public class ClickHouseBatchRunnable implements Runnable {
     private static final Logger log = LoggerFactory.getLogger(ClickHouseBatchRunnable.class);
-    private final ConcurrentLinkedQueue<List<ClickHouseStruct>> records;
+    private final LinkedBlockingQueue<List<ClickHouseStruct>> records;
 
     private final ClickHouseSinkConnectorConfig config;
 
@@ -57,7 +58,7 @@ public class ClickHouseBatchRunnable implements Runnable {
 
     private List<ClickHouseStruct> currentBatch = null;
 
-    public ClickHouseBatchRunnable(ConcurrentLinkedQueue<List<ClickHouseStruct>> records,
+    public ClickHouseBatchRunnable(LinkedBlockingQueue<List<ClickHouseStruct>> records,
                                    ClickHouseSinkConnectorConfig config,
                                    Map<String, String> topic2TableMap) {
         this.records = records;