From 8f16817b8d30c75c97a0570f6382bb0e2c5c4a38 Mon Sep 17 00:00:00 2001
From: Ludovic Boutros <lboutros@confluent.io>
Date: Thu, 7 Mar 2024 11:50:40 +0100
Subject: [PATCH] Fix: Disable mandatory field validation if validation is
 disabled

---
 .../com/splunk/kafka/connect/SplunkSinkConnector.java |  9 ++++-----
 .../splunk/kafka/connect/SplunkSinkConnecterTest.java | 11 +++++++++++
 2 files changed, 15 insertions(+), 5 deletions(-)

diff --git a/src/main/java/com/splunk/kafka/connect/SplunkSinkConnector.java b/src/main/java/com/splunk/kafka/connect/SplunkSinkConnector.java
index 3fb58e07..fa9f0410 100644
--- a/src/main/java/com/splunk/kafka/connect/SplunkSinkConnector.java
+++ b/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, "");
diff --git a/src/test/java/com/splunk/kafka/connect/SplunkSinkConnecterTest.java b/src/test/java/com/splunk/kafka/connect/SplunkSinkConnecterTest.java
index ba330fe4..d72cf01f 100644
--- a/src/test/java/com/splunk/kafka/connect/SplunkSinkConnecterTest.java
+++ b/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<>();