Remove default codec and require codec for sqs sink (#6486)

Signed-off-by: Taylor Gray <tylgry@amazon.com>

Author: graytaylor0

data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSink.java

@@ -34,7 +34,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
-import java.util.Map;
 import java.util.Collection;
 
 @Experimental
@@ -60,16 +59,12 @@ public SqsSink(final PluginSetting pluginSetting,
         sinkInitialized = false;
         final PluginModel codecConfiguration = sqsSinkConfig.getCodec();
         final PluginSetting codecPluginSettings;
-        if (codecConfiguration != null) {
-            String codecPluginName = codecConfiguration.getPluginName();
-            if (!codecPluginName.equals("json") && !codecPluginName.equals("ndjson")) {
-                throw new RuntimeException(String.format("Codec {} not supported.", codecPluginName));
-            }
-            codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
-                codecConfiguration.getPluginSettings());
-        } else {
-            codecPluginSettings = new PluginSetting("ndjson", Map.of());
+        String codecPluginName = codecConfiguration.getPluginName();
+        if (!codecPluginName.equals("json") && !codecPluginName.equals("ndjson")) {
+            throw new RuntimeException(String.format("Codec {} not supported.", codecPluginName));
         }
+        codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
+            codecConfiguration.getPluginSettings());
 
         AwsConfig awsConfig = sqsSinkConfig.getAwsConfig();
         final AwsCredentialsProvider awsCredentialsProvider = (awsConfig != null) ? awsCredentialsSupplier.getProvider(convertToCredentialOptions(awsConfig)) : awsCredentialsSupplier.getProvider(AwsCredentialsOptions.builder().build());

data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkConfig.java

@@ -28,6 +28,7 @@ public class SqsSinkConfig {
     private String queueUrl;
 
     @JsonProperty("codec")
+    @NotNull
     private PluginModel codec;
 
     @JsonProperty("threshold")
@@ -67,12 +68,5 @@ boolean isValidConfig() {
              return (deDupId == null);
         }
     }
-
-    @AssertTrue(message = "ndjson codec (default codec) doesn't support max events per message greater than 1")
-    boolean isValidCodecConfig() {
-        if ((codec == null || codec.getPluginName().equals("ndjson")) && thresholdConfig.getMaxEventsPerMessage() > 1)
-            return false;
-        return true;
-    }
 }
 

data-prepper-plugins/sqs-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkConfigTest.java

@@ -114,21 +114,6 @@ void TestValidFiFoQConfigs() throws Exception {
 
     }
 
-    @Test
-    void TestValidCodecConfig() throws Exception {
-        reflectivelySetField(sqsSinkConfig, "codec", null);
-        reflectivelySetField(sqsSinkConfig, "thresholdConfig", sqsThresholdConfig);
-        when(sqsThresholdConfig.getMaxEventsPerMessage()).thenReturn(2);
-        assertFalse(sqsSinkConfig.isValidCodecConfig());
-        when(sqsThresholdConfig.getMaxEventsPerMessage()).thenReturn(1);
-        assertTrue(sqsSinkConfig.isValidCodecConfig());
-        PluginModel codec = mock(PluginModel.class);
-        when(codec.getPluginName()).thenReturn("ndjson");
-        reflectivelySetField(sqsSinkConfig, "codec", codec);
-        when(sqsThresholdConfig.getMaxEventsPerMessage()).thenReturn(2);
-        assertFalse(sqsSinkConfig.isValidCodecConfig());
-    }
-
     private void reflectivelySetField(final SqsSinkConfig sqsSinkConfig, final String fieldName, final Object value) throws NoSuchFieldException, IllegalAccessException {
         final Field field = SqsSinkConfig.class.getDeclaredField(fieldName);
         try {

data-prepper-plugins/sqs-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkTest.java

@@ -230,20 +230,6 @@ void TestForDefaultCodec() {
         }
     }
 
-    @Test
-    void TestForNullCodec() {
-        when(sqsSinkConfig.getCodec()).thenReturn(null);
-        try(MockedStatic<SqsClientFactory> mockedStatic = mockStatic(SqsClientFactory.class)) {
-            mockedStatic.when(() -> SqsClientFactory.createSqsClient(any(Region.class),
-                            any(AwsCredentialsProvider.class)))
-                    .thenReturn(sqsClient);
-
-            SqsSink sqsSink = createObjectUnderTest();
-            sqsSink.doInitialize();
-            assertTrue(sqsSink.isReady());
-        }
-    }
-
     @Test
     void TestSinkOutputWithEvents() {
         try(MockedStatic<SqsClientFactory> mockedStatic = mockStatic(SqsClientFactory.class)) {