S3 DLQ: Add path style (deprecated) access

Signed-off-by: Simon ELBAZ <elbazsimon9@gmail.com>

Author: simonelbaz

data-prepper-plugins/failures-common/src/main/java/org/opensearch/dataprepper/plugins/dlq/s3/S3DlqWriterConfig.java

@@ -36,6 +36,7 @@ public class S3DlqWriterConfig {
     private static final String AWS_IAM_ROLE = "role";
     private static final String AWS_IAM = "iam";
     private static final String S3_PREFIX = "s3://";
+    private static final boolean FORCE_PATH_STYLE = false;
 
     @JsonProperty("bucket")
     @NotEmpty
@@ -50,6 +51,9 @@ public class S3DlqWriterConfig {
     @Size(min = 1, message = "region cannot be empty string")
     private String region = DEFAULT_AWS_REGION;
 
+    @JsonProperty("force_path_style")
+    private boolean forcePathStyle = FORCE_PATH_STYLE;
+
     @JsonProperty("sts_role_arn")
     @Size(min = 20, max = 2048, message = "sts_role_arn length should be between 1 and 2048 characters")
     private String stsRoleArn;
@@ -76,6 +80,10 @@ public String getKeyPathPrefix() {
         return keyPathPrefix;
     }
 
+    public boolean getForcePathStyle() {
+        return forcePathStyle;
+    }
+
     public Region getRegion() {
         return Region.of(region);
     }
@@ -135,6 +143,7 @@ private Arn getArn() {
     public S3Client getS3Client() {
         return S3Client.builder()
             .region(this.getRegion())
+            .forcePathStyle(this.getForcePathStyle())
             .credentialsProvider(this.getAwsCredentialsProvider())
             .overrideConfiguration(ClientOverrideConfiguration.builder()
                 .retryPolicy(RetryPolicy.builder().numRetries(MAX_NUMBER_OF_RETRIES).build())

data-prepper-plugins/failures-common/src/test/java/org/opensearch/dataprepper/plugins/dlq/s3/S3DlqWriterConfigTest.java

@@ -35,6 +35,11 @@ public void testDefaultKeyPathPrefix() {
         assertThat(new S3DlqWriterConfig().getKeyPathPrefix(), is(equalTo(null)));
     }
 
+    @Test
+    public void testDefaultForcePathStyle() {
+        assertThat(new S3DlqWriterConfig().getForcePathStyle(), is(equalTo(false)));
+    }
+
     @ParameterizedTest
     @ValueSource(strings = {"foobar", "arn:aws:es:us-west-2:123456789012:domain/bogus-domain",
         "arn:aws:iam::123456789012:group/bogus-group"})
@@ -62,6 +67,16 @@ public void getS3ClientWithValidStsRoleArn(final String stsRoleArn) throws NoSuc
         assertThat(s3Client, is(notNullValue()));
     }
 
+    @ParameterizedTest
+    @NullSource
+    @ValueSource(booleans = {false, true})
+    public void getS3ClientWithValidAccessStyle(final boolean forcePathStyle) throws NoSuchFieldException, IllegalAccessException {
+        final S3DlqWriterConfig config = new S3DlqWriterConfig();
+        reflectivelySetField(config, "force_path_style", forcePathStyle);
+        final S3Client s3Client = config.getS3Client();
+        assertThat(s3Client, is(notNullValue()));
+    }
+
     @ParameterizedTest
     @NullSource
     @ValueSource(strings = {"", "arn:aws:iam::123456789012:role/some-role"})

data-prepper-plugins/sns-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sns/dlq/DlqPushHandler.java

@@ -49,6 +49,8 @@ public class DlqPushHandler {
 
     private static final String KEY_PATH_PREFIX = "key_path_prefix";
 
+    private static final String FORCE_PATH_STYLE = "force_path_style";
+
     private String dlqFile;
 
     private String keyPathPrefix;
@@ -62,12 +64,13 @@ public DlqPushHandler(final String dlqFile,
                           final String bucket,
                           final String stsRoleArn,
                           final String awsRegion,
+                          final Boolean forcePathStyle,
                           final String dlqPathPrefix) {
         if(dlqFile != null) {
             this.dlqFile = dlqFile;
             this.objectWriter = new ObjectMapper().writer().withDefaultPrettyPrinter();
         }else{
-            this.dlqProvider = getDlqProvider(pluginFactory,bucket,stsRoleArn,awsRegion,dlqPathPrefix);
+            this.dlqProvider = getDlqProvider(pluginFactory,bucket,stsRoleArn,awsRegion,forcePathStyle,dlqPathPrefix);
         }
     }
 
@@ -117,11 +120,13 @@ private  DlqProvider getDlqProvider(final PluginFactory pluginFactory,
                                         final String bucket,
                                         final String stsRoleArn,
                                         final String awsRegion,
+                                        final Boolean forcePathStyle,
                                         final String dlqPathPrefix) {
         final Map<String, Object> props = new HashMap<>();
         props.put(BUCKET, bucket);
         props.put(ROLE_ARN, stsRoleArn);
         props.put(REGION, awsRegion);
+        props.put(FORCE_PATH_STYLE, forcePathStyle);
         this.keyPathPrefix = StringUtils.isEmpty(dlqPathPrefix) ? dlqPathPrefix : enforceDefaultDelimiterOnKeyPathPrefix(dlqPathPrefix);
         props.put(KEY_PATH_PREFIX, dlqPathPrefix);
         final PluginSetting dlqPluginSetting = new PluginSetting(S3_PLUGIN_NAME, props);

data-prepper-plugins/sns-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sns/dlq/DlqPushHandlerTest.java

@@ -39,6 +39,11 @@ class DlqPushHandlerTest {
 
     private static final String KEY_PATH_PREFIX_VALUE = "dlq/";
 
+    private static final String FORCE_PATH_STYLE = "force_path_style";
+
+    private static final boolean FORCE_PATH_STYLE_TRUE = true;
+    private static final boolean FORCE_PATH_STYLE_FALSE = false;
+
     private static final String PIPELINE_NAME = "log-pipeline";
 
     private static final String DLQ_FILE = "local_dlq_file";
@@ -84,6 +89,48 @@ void perform_for_dlq_s3_success() throws IOException {
         verify(dlqWriter).write(anyList(), anyString(), anyString());
     }
 
+    @Test
+    void perform_for_dlq_s3_success_forcepathstyle_true() throws IOException {
+        Map<String, Object> props = new HashMap<>();
+        props.put(BUCKET,BUCKET_VALUE);
+        props.put(KEY_PATH_PREFIX,KEY_PATH_PREFIX_VALUE);
+        props.put(FORCE_PATH_STYLE,FORCE_PATH_STYLE_TRUE);
+
+        when(pluginFactory.loadPlugin(any(Class.class), any(PluginSetting.class))).thenReturn(dlqProvider);
+
+        when(dlqProvider.getDlqWriter(anyString())).thenReturn(Optional.of(dlqWriter));
+        doNothing().when(dlqWriter).write(anyList(), anyString(), anyString());
+        SnsSinkFailedDlqData failedDlqData = new SnsSinkFailedDlqData("topic","message",0);
+        dlqPushHandler = new DlqPushHandler(null,pluginFactory, BUCKET_VALUE, ROLE, REGION,KEY_PATH_PREFIX_VALUE,FORCE_PATH_STYLE_TRUE);
+
+        PluginSetting pluginSetting = new PluginSetting(S3_PLUGIN_NAME, props);
+        pluginSetting.setPipelineName(PIPELINE_NAME);
+        dlqPushHandler.perform(pluginSetting, failedDlqData);
+        Assertions.assertNotNull(pluginFactory);
+        verify(dlqWriter).write(anyList(), anyString(), anyString());
+    }
+
+    @Test
+    void perform_for_dlq_s3_success_forcepathstyle_false() throws IOException {
+        Map<String, Object> props = new HashMap<>();
+        props.put(BUCKET,BUCKET_VALUE);
+        props.put(KEY_PATH_PREFIX,KEY_PATH_PREFIX_VALUE);
+        props.put(FORCE_PATH_STYLE,FORCE_PATH_STYLE_FALSE);
+
+        when(pluginFactory.loadPlugin(any(Class.class), any(PluginSetting.class))).thenReturn(dlqProvider);
+
+        when(dlqProvider.getDlqWriter(anyString())).thenReturn(Optional.of(dlqWriter));
+        doNothing().when(dlqWriter).write(anyList(), anyString(), anyString());
+        SnsSinkFailedDlqData failedDlqData = new SnsSinkFailedDlqData("topic","message",0);
+        dlqPushHandler = new DlqPushHandler(null,pluginFactory, BUCKET_VALUE, ROLE, REGION,KEY_PATH_PREFIX_VALUE,FORCE_PATH_STYLE_FALSE);
+
+        PluginSetting pluginSetting = new PluginSetting(S3_PLUGIN_NAME, props);
+        pluginSetting.setPipelineName(PIPELINE_NAME);
+        dlqPushHandler.perform(pluginSetting, failedDlqData);
+        Assertions.assertNotNull(pluginFactory);
+        verify(dlqWriter).write(anyList(), anyString(), anyString());
+    }
+
 
     @Test
     void perform_for_dlq_local_file_success(){