Fix broken DefaultKafkaClusterConfigSupplier get API (opensearch-proj…

…ect#3529) Signed-off-by: Kondaka <[email protected]>
daixba · Oct 19, 2023 · c63dea3 · c63dea3
1 parent 61efeae
commit c63dea3
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 5 deletions.
diff --git a/...org/opensearch/dataprepper/plugins/kafka/extension/DefaultKafkaClusterConfigSupplier.java b/...org/opensearch/dataprepper/plugins/kafka/extension/DefaultKafkaClusterConfigSupplier.java
@@ -10,6 +10,7 @@
 import org.opensearch.dataprepper.plugins.kafka.configuration.EncryptionConfig;
 
 import java.util.List;
+import java.util.Objects;
 
 public class DefaultKafkaClusterConfigSupplier implements KafkaClusterConfigSupplier {
     private final KafkaClusterConfig kafkaClusterConfig;
@@ -19,21 +20,21 @@ public DefaultKafkaClusterConfigSupplier(KafkaClusterConfig kafkaClusterConfig)
 
     @Override
     public List<String> getBootStrapServers() {
-        return kafkaClusterConfig.getBootStrapServers();
+        return Objects.nonNull(kafkaClusterConfig) ? kafkaClusterConfig.getBootStrapServers() : null;
     }
 
     @Override
     public AuthConfig getAuthConfig() {
-        return kafkaClusterConfig.getAuthConfig();
+        return Objects.nonNull(kafkaClusterConfig) ? kafkaClusterConfig.getAuthConfig() : null;
     }
 
     @Override
     public AwsConfig getAwsConfig() {
-        return kafkaClusterConfig.getAwsConfig();
+        return Objects.nonNull(kafkaClusterConfig) ? kafkaClusterConfig.getAwsConfig() : null;
     }
 
     @Override
     public EncryptionConfig getEncryptionConfig() {
-        return kafkaClusterConfig.getEncryptionConfig();
+        return Objects.nonNull(kafkaClusterConfig) ? kafkaClusterConfig.getEncryptionConfig() : null;
     }
 }
diff --git a/...opensearch/dataprepper/plugins/kafka/extension/DefaultKafkaClusterConfigSupplierTest.java b/...opensearch/dataprepper/plugins/kafka/extension/DefaultKafkaClusterConfigSupplierTest.java
@@ -30,7 +30,7 @@ private DefaultKafkaClusterConfigSupplier createObjectUnderTest() {
     }
 
     @Test
-    void test_getters() {
+    void testGetters() {
         final List<String> bootstrapServers = List.of("localhost:9092");
         final AuthConfig authConfig = mock(AuthConfig.class);
         final AwsConfig awsConfig = mock(AwsConfig.class);
@@ -45,4 +45,14 @@ void test_getters() {
         assertThat(defaultKafkaClusterConfigSupplier.getAwsConfig(), equalTo(awsConfig));
         assertThat(defaultKafkaClusterConfigSupplier.getEncryptionConfig(), equalTo(encryptionConfig));
     }
+
+    @Test
+    void testGettersWithNullClusterConfig() {
+        DefaultKafkaClusterConfigSupplier defaultKafkaClusterConfigSupplier = 
+		new DefaultKafkaClusterConfigSupplier(null);
+        assertThat(defaultKafkaClusterConfigSupplier.getBootStrapServers(), equalTo(null));
+        assertThat(defaultKafkaClusterConfigSupplier.getAuthConfig(), equalTo(null));
+        assertThat(defaultKafkaClusterConfigSupplier.getAwsConfig(), equalTo(null));
+        assertThat(defaultKafkaClusterConfigSupplier.getEncryptionConfig(), equalTo(null));
+    }
 }