Merge pull request #387 from splunk/queue_capacity

Make queue capacity configurable

Author: VihasMakwana

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,7 +317,8 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         regex = getString(REGEX_CONF);
         timestampFormat = getString(TIMESTAMP_FORMAT_CONF).trim();
         validateRegexForTimestamp(regex);
-      
+        queueCapacity = getInt(QUEUE_CAPACITY_CONF);
+        validateQueueCapacity(queueCapacity);
     }
 
 
@@ -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;
     }
 
@@ -550,6 +558,12 @@ private void validateRegexForTimestamp(String regex) {
         }
     }
 
+    private void validateQueueCapacity(int queueCapacity) {
+        if (queueCapacity <= 0) {
+            throw new ConfigException("queue capacity should be greater than " + queueCapacity);
+        }
+    }
+
     private static boolean getNamedGroupCandidates(String regex) {
         Matcher m = Pattern.compile("\\(\\?<([a-zA-Z][a-zA-Z0-9]*)>").matcher(regex);
         while (m.find()) {

src/test/java/com/splunk/kafka/connect/SplunkSinkConnecterTest.java

@@ -261,6 +261,34 @@ public void testInvalidSplunkConfigurationsWithValidationEnabled() {
         Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
     }
 
+    @Test
+    public void testValidQueueCapacity() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SplunkSinkConnector connector = new SplunkSinkConnector();
+        configs.put("splunk.hec.concurrent.queue.capacity", "100");
+        configs.put("topics", "b");
+        configs.put("splunk.indexes", "b");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        clientInstance.client.setResponse(CloseableHttpClientMock.success);
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertDoesNotThrow(()->connector.validate(configs));
+    }
+
+    @Test
+    public void testInvalidQueueCapacity() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SplunkSinkConnector connector = new SplunkSinkConnector();
+        configs.put("splunk.hec.concurrent.queue.capacity", "-1");
+        configs.put("topics", "b");
+        configs.put("splunk.indexes", "b");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        clientInstance.client.setResponse(CloseableHttpClientMock.success);
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
+    }
+
     private void addNecessaryConfigs(Map<String, String> configs) {
         configs.put(URI_CONF, TEST_URI);
         configs.put(TOKEN_CONF, "blah");