Addressed review comments

Signed-off-by: Krishna Kondaka <[email protected]>

Author: kkondaka

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsSink.java

@@ -55,7 +55,13 @@ public CloudWatchLogsSink(final PluginSetting pluginSetting,
                 thresholdConfig.getMaxEventSizeBytes(),
                 thresholdConfig.getMaxRequestSizeBytes(),thresholdConfig.getLogSendInterval());
 
+        if (awsConfig == null && awsCredentialsSupplier == null) {
+            throw new RuntimeException("Missing awsConfig and awsCredentialsSupplier");
+        }
         CloudWatchLogsClient cloudWatchLogsClient = CloudWatchLogsClientFactory.createCwlClient(awsConfig, awsCredentialsSupplier);
+        if (cloudWatchLogsClient == null) {
+            throw new RuntimeException("cloudWatchLogsClient is null");
+        }
 
         BufferFactory bufferFactory = null;
         if (cloudWatchLogsSinkConfig.getBufferType().equals("in_memory")) {

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsClientFactory.java

@@ -10,6 +10,7 @@
 import org.opensearch.dataprepper.plugins.sink.cloudwatch_logs.config.AwsConfig;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
+import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
 
 /**
@@ -27,11 +28,14 @@ private CloudWatchLogsClientFactory() {
      * @return CloudWatchLogsClient used to interact with CloudWatch Logs services.
      */
     public static CloudWatchLogsClient createCwlClient(final AwsConfig awsConfig, final AwsCredentialsSupplier awsCredentialsSupplier) {
-        final AwsCredentialsOptions awsCredentialsOptions = convertToCredentialOptions(awsConfig);
-        final AwsCredentialsProvider awsCredentialsProvider = awsCredentialsSupplier.getProvider(awsCredentialsOptions);
+        final AwsCredentialsProvider awsCredentialsProvider = awsConfig != null ? awsCredentialsSupplier.getProvider(convertToCredentialOptions(awsConfig)) : awsCredentialsSupplier.getProvider(AwsCredentialsOptions.defaultOptionsWithDefaultCredentialsProvider());
+        Region region = awsConfig != null ? awsConfig.getAwsRegion() : awsCredentialsSupplier.getDefaultRegion().get();
 
+        if (awsCredentialsProvider == null || region == null) {
+            return null;
+        }
         return CloudWatchLogsClient.builder()
-                .region(awsConfig.getAwsRegion())
+                .region(region)
                 .credentialsProvider(awsCredentialsProvider)
                 .overrideConfiguration(createOverrideConfiguration()).build();
     }

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsService.java

@@ -36,7 +36,6 @@ public class CloudWatchLogsService {
     private final CloudWatchLogsDispatcher cloudWatchLogsDispatcher;
     private final Buffer buffer;
     private final CloudWatchLogsLimits cloudWatchLogsLimits;
-    //private List<EventHandle> bufferedEventHandles;
     private final SinkStopWatch sinkStopWatch;
     private final ReentrantLock processLock;
     private final DlqPushHandler dlqPushHandler;
@@ -48,8 +47,6 @@ public CloudWatchLogsService(final Buffer buffer,
         this.buffer = buffer;
         this.cloudWatchLogsLimits = cloudWatchLogsLimits;
 
-        //bufferedEventHandles = new ArrayList<>();
-
         processLock = new ReentrantLock();
         sinkStopWatch = new SinkStopWatch();
 

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/config/CloudWatchLogsSinkConfig.java

@@ -17,7 +17,6 @@ public class CloudWatchLogsSinkConfig {
     public static final String DEFAULT_BUFFER_TYPE = "in_memory";
 
     @JsonProperty("aws")
-    @NotNull
     @Valid
     private AwsConfig awsConfig;
 

data-prepper-plugins/cloudwatch-logs/src/test/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsSinkTest.java

@@ -26,6 +26,7 @@
 import java.util.Collection;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.mock;
@@ -100,6 +101,19 @@ void WHEN_sink_is_initialized_THEN_sink_is_ready_returns_true() {
         }
     }
 
+    @Test
+    void WHEN_awsConfig_and_awsCredentialsSupplier_null_THEN_should_throw() {
+        mockCredentialSupplier = null;
+        when(mockCloudWatchLogsSinkConfig.getAwsConfig()).thenReturn(null);
+        try(MockedStatic<CloudWatchLogsClientFactory> mockedStatic = mockStatic(CloudWatchLogsClientFactory.class)) {
+            mockedStatic.when(() -> CloudWatchLogsClientFactory.createCwlClient(any(AwsConfig.class),
+                            any(AwsCredentialsSupplier.class)))
+                    .thenReturn(mockClient);
+
+            assertThrows(RuntimeException.class, ()-> getTestCloudWatchSink());
+        }
+    }
+
     @Test
     void WHEN_given_sample_empty_records_THEN_records_are_processed() {
         try(MockedStatic<CloudWatchLogsClientFactory> mockedStatic = mockStatic(CloudWatchLogsClientFactory.class)) {

data-prepper-plugins/cloudwatch-logs/src/test/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsClientFactoryTest.java

@@ -19,12 +19,16 @@
 import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
 import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClientBuilder;
 
+
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.mockStatic;
@@ -34,22 +38,41 @@ class CloudWatchLogsClientFactoryTest {
     private AwsConfig mockAwsConfig;
     private AwsCredentialsSupplier mockAwsCredentialsSupplier;
     private AwsCredentialsOptions mockAwsCredentialsOptions;
+    private AwsCredentialsProvider mockAwsCredentialsProvider;
 
     @BeforeEach
     void setUp() {
         mockAwsConfig = mock(AwsConfig.class);
         mockAwsCredentialsSupplier = mock(AwsCredentialsSupplier.class);
         mockAwsCredentialsOptions = mock(AwsCredentialsOptions.class);
+        mockAwsCredentialsProvider = mock(AwsCredentialsProvider.class);
         when(mockAwsConfig.getAwsRegion()).thenReturn(Region.US_EAST_1);
+        when(mockAwsCredentialsSupplier.getDefaultRegion()).thenReturn(Optional.of(Region.US_EAST_1));
     }
 
     @Test
     void GIVEN_default_credentials_SHOULD_return_non_null_client() {
+        when(mockAwsCredentialsSupplier.getProvider(any())).thenReturn(mockAwsCredentialsProvider);
         final CloudWatchLogsClient cloudWatchLogsClientToTest = CloudWatchLogsClientFactory.createCwlClient(mockAwsConfig, mockAwsCredentialsSupplier);
 
         assertNotNull(cloudWatchLogsClientToTest);
     }
 
+    @Test
+    void GIVEN_default_credentials_with_no_provider_SHOULD_return_null_client() {
+        final CloudWatchLogsClient cloudWatchLogsClientToTest = CloudWatchLogsClientFactory.createCwlClient(mockAwsConfig, mockAwsCredentialsSupplier);
+
+        assertNull(cloudWatchLogsClientToTest);
+    }
+
+    @Test
+    void GIVEN_default_credentials_with_no_region_SHOULD_return_null_client() {
+        when(mockAwsConfig.getAwsRegion()).thenReturn(null);
+        final CloudWatchLogsClient cloudWatchLogsClientToTest = CloudWatchLogsClientFactory.createCwlClient(mockAwsConfig, mockAwsCredentialsSupplier);
+
+        assertNull(cloudWatchLogsClientToTest);
+    }
+
     @Test
     void GIVEN_valid_credential_and_aws_parameters_SHOULD_generate_valid_provider_and_options() {
         final String stsRoleArn = UUID.randomUUID().toString();