Merge pull request #424 from ludovic-boutros/fix-421-validate-endpoint-failure

Fix: Disable mandatory field validation if validation is disabled

Author: VihasMakwana

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

@@ -15,9 +15,6 @@
  */
 package com.splunk.kafka.connect;
 
-import static com.splunk.kafka.connect.SplunkSinkConnectorConfig.KERBEROS_KEYTAB_PATH_CONF;
-import static com.splunk.kafka.connect.SplunkSinkConnectorConfig.KERBEROS_USER_PRINCIPAL_CONF;
-
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -53,6 +50,8 @@
 import com.splunk.hecclient.JsonEvent;
 import com.splunk.hecclient.JsonEventBatch;
 
+import static com.splunk.kafka.connect.SplunkSinkConnectorConfig.*;
+
 public final class SplunkSinkConnector extends SinkConnector {
     private static final Logger log = LoggerFactory.getLogger(SplunkSinkConnector.class);
     private Map<String, String> taskConfig;
@@ -147,10 +146,10 @@ private static String[] split(String data, String sep) {
 
 
     private void validateSplunkConfigurations(final Map<String, String> configs) throws ConfigException {
-        SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(configs);
-        if (connectorConfig.disableValidation) {
+        if (configs.containsKey(DISABLE_VALIDATION) && Boolean.parseBoolean(configs.get(DISABLE_VALIDATION))) {
             return;
         }
+        SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(configs);
         String[] indexes = split(connectorConfig.indexes, ",");
         if(indexes == null || indexes.length == 0) {
             preparePayloadAndExecuteRequest(connectorConfig, "");

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

@@ -248,6 +248,17 @@ public void testInvalidSplunkConfigurationsWithValidationDisabled() {
         Assertions.assertDoesNotThrow(()->connector.validate(configs));
     }
 
+    @Test
+    public void testInvalidSplunkConfigurationWithMandatoryFieldMissingWithValidationDisabled() {
+        final Map<String, String> configs = new HashMap<>();
+        SplunkSinkConnector connector = new SplunkSinkConnector();
+        configs.put("splunk.validation.disable", "true");
+        configs.put("topics", "b");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        clientInstance.client.setResponse(CloseableHttpClientMock.EXCEPTION);
+        Assertions.assertDoesNotThrow(()->connector.validate(configs));
+    }
+
     @Test
     public void testInvalidSplunkConfigurationsWithValidationEnabled() {
         final Map<String, String> configs = new HashMap<>();