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";
@@ -257,6 +260,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 +316,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         regex = getString(REGEX_CONF);
         timestampFormat = getString(TIMESTAMP_FORMAT_CONF).trim();
         validateRegexForTimestamp(regex);
+        queueCapacity = getInt(QUEUE_CAPACITY_CONF);
 
     }
 
@@ -385,6 +390,7 @@ public HecConfig getHecConfig() {
               .setHasCustomTrustStore(hasTrustStorePath)
               .setKerberosPrincipal(kerberosUserPrincipal)
               .setKerberosKeytabPath(kerberosKeytabPath);
+              .setConcurrentHecQueueCapacity(queueCapacity);
         return config;
     }