Fix s3 folder scan depth for RDS source when partition prefix is not set

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>

Author: divbok

data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/transformer/DynamicConfigTransformer.java

@@ -427,13 +427,19 @@ public String calculateDepth(String s3Prefix) {
         return Integer.toString(getDepth(s3Prefix, 4));
     }
 
+    protected String getSourceCoordinationIdentifier() {
+        return System.getenv(SOURCE_COORDINATION_IDENTIFIER_ENVIRONMENT_VARIABLE);
+    }
+
     /**
      * Calculate s3 folder scan depth for RDS source pipeline
      * @param s3Prefix: s3 prefix defined in the source configuration
      * @return s3 folder scan depth
      */
     public String calculateDepthForRdsSource(String s3Prefix) {
-        return Integer.toString(getDepth(s3Prefix, 3));
+        String envSourceCoordinationIdentifier = getSourceCoordinationIdentifier();
+        int baseDepth = envSourceCoordinationIdentifier != null ? 3 : 2;
+        return Integer.toString(getDepth(s3Prefix, baseDepth));
     }
 
     private int getDepth(String s3Prefix, int baseDepth) {

data-prepper-pipeline-parser/src/test/java/org/opensearch/dataprepper/pipeline/parser/transformer/DynamicConfigTransformerTest.java

@@ -37,8 +37,12 @@
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 
 class DynamicConfigTransformerTest {
@@ -456,4 +460,23 @@ void test_overlay_directive_overrides_existing_fields() throws Exception {
         assertThat(resultOs.get("script").get("source").asText()).isEqualTo("ctx._source.merge(params.doc)");
         assertThat(resultOs.get("script").has("custom_field")).isFalse();
     }
+
+    @Test
+    void test_calculateDepthForRdsSource_without_source_coordination_identifier() {
+        String mockPrefix = "my-bucket/path";
+        DynamicConfigTransformer transformer = spy(new DynamicConfigTransformer(mock(RuleEvaluator.class)));
+        doReturn(null).when(transformer).getSourceCoordinationIdentifier();
+        String result = transformer.calculateDepthForRdsSource(mockPrefix);
+        assertThat(result, equalTo("4"));
+    }
+
+    @Test
+    void test_calculateDepthForRdsSource_with_source_coordination_identifier() {
+        String mockPrefix = "my-bucket/path";
+        DynamicConfigTransformer transformer = spy(new DynamicConfigTransformer(mock(RuleEvaluator.class)));
+        doReturn("testValue").when(transformer).getSourceCoordinationIdentifier();
+        String result = transformer.calculateDepthForRdsSource(mockPrefix);
+        assertThat(result, equalTo("5"));
+    }
+
 }