Make queue capacity configurable

Author: Vihas Splunk

src/main/java/com/splunk/hecclient/ConcurrentHec.java

@@ -37,7 +37,7 @@ public ConcurrentHec(int numberOfThreads, boolean useAck, HecConfig config, Poll
     }
 
     public ConcurrentHec(int numberOfThreads, boolean useAck, HecConfig config, PollerCallback cb, LoadBalancerInf loadBalancer) {
-        batches = new LinkedBlockingQueue<>(100);
+        batches = new LinkedBlockingQueue<>(config.getConcurrentHecQueueCapacity());
         ThreadFactory e = (Runnable r) -> new Thread(r, "Concurrent-HEC-worker");
         executorService = Executors.newFixedThreadPool(numberOfThreads, e);
         initHec(numberOfThreads, useAck, config, cb, loadBalancer);

src/main/java/com/splunk/hecclient/HecConfig.java

@@ -39,6 +39,7 @@ public final class HecConfig {
     private int lbPollInterval = 120; // in seconds
     private String kerberosPrincipal;
     private String kerberosKeytabPath;
+    private int concurrentHecQueueCapacity = 100;
 
     public HecConfig(List<String> uris, String token) {
         this.uris = uris;
@@ -101,6 +102,10 @@ public int getBackoffThresholdSeconds() {
         return backoffThresholdSeconds;
     }
 
+    public int getConcurrentHecQueueCapacity() {
+        return concurrentHecQueueCapacity;
+    }
+
     public boolean getHasCustomTrustStore() { return hasCustomTrustStore; }
 
     public String getTrustStorePath() { return trustStorePath; }
@@ -207,6 +212,11 @@ public HecConfig setKerberosKeytabPath(String kerberosKeytabPath) {
         return this;
     }
 
+    public HecConfig setConcurrentHecQueueCapacity(int concurrentHecQueueCapacity) {
+        this.concurrentHecQueueCapacity = concurrentHecQueueCapacity;
+        return this;
+    }
+
     public boolean kerberosAuthEnabled() {
         return !kerberosPrincipal().isEmpty();
     }

src/main/java/com/splunk/kafka/connect/SplunkSinkConnectorConfig.java

@@ -55,6 +55,9 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
     static final String SOCKET_TIMEOUT_CONF = "splunk.hec.socket.timeout"; // seconds
     static final String SSL_VALIDATE_CERTIFICATES_CONF = "splunk.hec.ssl.validate.certs";
     static final String ENABLE_COMPRESSSION_CONF = "splunk.hec.enable.compression";
+    // only applicable when "splunk.hec.threads" > 1
+    static final String QUEUE_CAPACITY_CONF = "splunk.hec.concurrent.queue.capacity";
+
     // Acknowledgement Parameters
     // Use Ack
     static final String ACK_CONF = "splunk.hec.ack.enabled";
@@ -192,6 +195,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
     static final String HEADER_SOURCETYPE_DOC = "Header to use for Splunk Header Sourcetype";
     static final String HEADER_HOST_DOC = "Header to use for Splunk Header Host";
 
+    static final String QUEUE_CAPACITY_DOC = "This setting controls the queue capacity for concurrency";
     // Load Balancer
     static final String LB_POLL_INTERVAL_DOC = "This setting controls the load balancer polling interval. By default, "
             + "this setting is 120 seconds.";
@@ -257,6 +261,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
     final boolean enableTimestampExtraction;
     final String regex;
     final String timestampFormat;
+    final int queueCapacity;
 
     SplunkSinkConnectorConfig(Map<String, String> taskConfig) {
         super(conf(), taskConfig);
@@ -312,6 +317,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         regex = getString(REGEX_CONF);
         timestampFormat = getString(TIMESTAMP_FORMAT_CONF).trim();
         validateRegexForTimestamp(regex);
+        queueCapacity = getInt(QUEUE_CAPACITY_CONF);
 
     }
 
@@ -360,7 +366,8 @@ public static ConfigDef conf() {
                 .define(KERBEROS_KEYTAB_PATH_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, KERBEROS_KEYTAB_LOCATION_DOC)
                 .define(ENABLE_TIMESTAMP_EXTRACTION_CONF, ConfigDef.Type.BOOLEAN,  false , ConfigDef.Importance.MEDIUM, ENABLE_TIMESTAMP_EXTRACTION_DOC)
                 .define(REGEX_CONF, ConfigDef.Type.STRING,  "" , ConfigDef.Importance.MEDIUM, REGEX_DOC)
-                .define(TIMESTAMP_FORMAT_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, TIMESTAMP_FORMAT_DOC);         
+                .define(TIMESTAMP_FORMAT_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, TIMESTAMP_FORMAT_DOC)
+                .define(QUEUE_CAPACITY_CONF, ConfigDef.Type.INT, 100, ConfigDef.Importance.LOW, QUEUE_CAPACITY_DOC);       
     }
 
     /**
@@ -384,7 +391,8 @@ public HecConfig getHecConfig() {
               .setTrustStorePassword(trustStorePassword)
               .setHasCustomTrustStore(hasTrustStorePath)
               .setKerberosPrincipal(kerberosUserPrincipal)
-              .setKerberosKeytabPath(kerberosKeytabPath);
+              .setKerberosKeytabPath(kerberosKeytabPath)
+              .setConcurrentHecQueueCapacity(queueCapacity);
         return config;
     }