Add test cases

Author: Vihas Splunk

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

@@ -318,7 +318,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         timestampFormat = getString(TIMESTAMP_FORMAT_CONF).trim();
         validateRegexForTimestamp(regex);
         queueCapacity = getInt(QUEUE_CAPACITY_CONF);
-      
+        validateQueueCapacity(queueCapacity);
     }
 
 
@@ -558,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");