Support continuous scan mode on RDS source

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

@@ -50,6 +50,7 @@ public class TestConfigurationProvider {
     public static final String USER_CONFIG_TRANSFORMATION_DOCUMENTDB_SUBPIPELINES_ROUTES_CONFIG_FILE = "src/test/resources/transformation/userConfig/documentdb-subpipelines-routes-userconfig.yaml";
     public static final String USER_CONFIG_TRANSFORMATION_DOCUMENTDB_FUNCTION_CONFIG_FILE = "src/test/resources/transformation/userConfig/documentdb-function-userconfig.yaml";
     public static final String USER_CONFIG_TRANSFORMATION_DOCUMENTDB_SCAN_MODE_CONFIG_FILE = "src/test/resources/transformation/userConfig/documentdb-scan-mode-userconfig.yaml";
+    public static final String USER_CONFIG_TRANSFORMATION_RDS_SCAN_MODE_CONFIG_FILE = "src/test/resources/transformation/userConfig/rds-scan-mode-userconfig.yaml";
 
     public static final String RULES_TRANSFORMATION_DOCDB1_CONFIG_FILE = "src/test/resources/transformation/rules/documentdb1-rule.yaml";
     public static final String RULES_TRANSFORMATION_DOCUMENTDB_CONFIG_FILE = "src/test/resources/transformation/rules/documentdb-rule.yaml";

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

@@ -576,6 +576,52 @@ void test_invokeMethod_throws_when_no_provider_has_the_method() {
                 "Expected 'could not find' message, got: " + exception.getMessage());
     }
 
+    @Test
+    void test_transformation_with_rds_scan_mode_continuous_propagates_idle_scan_mode() throws IOException {
+        // Use the REAL rds template from the rds-source plugin
+        String userConfig = TestConfigurationProvider.USER_CONFIG_TRANSFORMATION_RDS_SCAN_MODE_CONFIG_FILE;
+        String realTemplatePath = "../data-prepper-plugins/rds-source/src/main/resources/org/opensearch/dataprepper/transforms/templates/rds-template.yaml";
+        String ruleFilePath = "src/test/resources/transformation/rules/rds-rule.yaml";
+        String pluginName = "rds";
+
+        PipelineConfigurationReader pipelineConfigurationReader = new PipelineConfigurationFileReader(userConfig);
+        final PipelinesDataflowModelParser pipelinesDataflowModelParser =
+                new PipelinesDataflowModelParser(pipelineConfigurationReader);
+
+        TransformersFactory transformersFactory = mock(TransformersFactory.class);
+
+        InputStream ruleStream = new FileInputStream(ruleFilePath);
+        InputStream templateStream = new FileInputStream(realTemplatePath);
+        RuleStream ruleInputStream = new RuleStream(Paths.get(ruleFilePath).getFileName().toString(), ruleStream);
+
+        List<RuleStream> ruleStreams = Collections.singletonList(ruleInputStream);
+        when(transformersFactory.loadRules()).thenReturn(ruleStreams);
+        when(transformersFactory.getPluginTemplateFileStream(pluginName)).thenReturn(templateStream);
+
+        RuleEvaluator ruleEvaluator = new RuleEvaluator(transformersFactory);
+
+        PipelinesDataFlowModel pipelinesDataFlowModel = pipelinesDataflowModelParser.parseConfiguration();
+        PipelineConfigurationTransformer transformer = new DynamicConfigTransformer(ruleEvaluator);
+        PipelinesDataFlowModel transformedModel = transformer.transformConfiguration(pipelinesDataFlowModel);
+        String transformedYaml = yamlMapper.writeValueAsString(transformedModel);
+
+        Map<String, Object> transformedMap = yamlMapper.readValue(transformedYaml, Map.class);
+
+        // The s3 sub-pipeline should be named "rds-pipeline-s3"
+        Map<String, Object> s3Pipeline = (Map<String, Object>) transformedMap.get("rds-pipeline-s3");
+        assertThat(s3Pipeline).as("s3 sub-pipeline should exist").isNotNull();
+
+        Map<String, Object> source = (Map<String, Object>) s3Pipeline.get("source");
+        Map<String, Object> s3Source = (Map<String, Object>) source.get("s3");
+        Map<String, Object> scan = (Map<String, Object>) s3Source.get("scan");
+        Map<String, Object> folderPartitions = (Map<String, Object>) scan.get("folder_partitions");
+
+        assertThat(folderPartitions).as("folder_partitions should exist").isNotNull();
+        assertThat(folderPartitions.get("idle_scan_mode"))
+                .as("idle_scan_mode should be 'continuous' when scan_mode is set to continuous")
+                .isEqualTo("continuous");
+    }
+
     public static class ValidProviderNoAnnotation implements org.opensearch.dataprepper.model.plugin.PipelineTransformFunctionProvider {
         public static String unannotatedMethod(String input) {
             return input;

data-prepper-pipeline-parser/src/test/resources/transformation/userConfig/rds-scan-mode-userconfig.yaml

@@ -0,0 +1,46 @@
+#
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+
+rds-pipeline:
+  source:
+    rds:
+      db_identifier: "my-rds-cluster"
+      cluster: true
+      engine: "mysql"
+      database: "mydb"
+      acknowledgments: true
+      authentication:
+        username: ${{aws_secrets:secret:username}}
+        password: ${{aws_secrets:secret:password}}
+      aws:
+        sts_role_arn: "arn:aws:iam::123456789012:role/MyRole"
+        region: "us-east-1"
+      s3_bucket: "my-staging-bucket"
+      s3_region: "us-east-1"
+      s3_prefix: "osis-rds"
+      scan_mode: "continuous"
+      stream: true
+  sink:
+    - opensearch:
+        hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"]
+        index: "my-index"
+        index_type: custom
+        document_id: "${getMetadata(\"primary_key\")}"
+        action: "${getMetadata(\"opensearch_action\")}"
+        aws:
+          sts_role_arn: "arn:aws:iam::123456789012:role/MyRole"
+          region: "us-east-1"
+
+extension:
+  aws:
+    secrets:
+      secret:
+        secret_id: "my-rds-secret"
+        region: "us-east-1"
+        refresh_interval: PT1H

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfig.java

@@ -14,6 +14,7 @@
 import jakarta.validation.constraints.NotNull;
 import org.opensearch.dataprepper.plugins.source.rds.configuration.AwsAuthenticationConfig;
 import org.opensearch.dataprepper.plugins.source.rds.configuration.EngineType;
+import org.opensearch.dataprepper.plugins.source.rds.configuration.ScanMode;
 import org.opensearch.dataprepper.plugins.source.rds.configuration.ExportConfig;
 import org.opensearch.dataprepper.plugins.source.rds.configuration.JoinConfig;
 import org.opensearch.dataprepper.plugins.source.rds.configuration.TableFilterConfig;
@@ -104,6 +105,9 @@ public class RdsSourceConfig {
     @JsonProperty("disable_s3_read_for_leader")
     private boolean disableS3ReadForLeader = false;
 
+    @JsonProperty("scan_mode")
+    private ScanMode scanMode = ScanMode.IDLE_OPTIMIZED;
+
     @JsonProperty("joins")
     @Valid
     private JoinConfig joinConfig;
@@ -192,6 +196,10 @@ public JoinConfig getJoinConfig() {
         return joinConfig;
     }
 
+    public ScanMode getScanMode() {
+        return this.scanMode;
+    }
+
     public AuthenticationConfig getAuthenticationConfig() {
         return this.authenticationConfig;
     }

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/configuration/ScanMode.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins.source.rds.configuration;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public enum ScanMode {
+    IDLE_OPTIMIZED("idle_optimized"),
+    CONTINUOUS("continuous");
+
+    private static final Map<String, ScanMode> NAMES_MAP = Arrays.stream(ScanMode.values())
+            .collect(Collectors.toMap(ScanMode::toString, v -> v));
+
+    private final String name;
+
+    ScanMode(final String name) {
+        this.name = name;
+    }
+
+    @JsonValue
+    public String toString() {
+        return this.name;
+    }
+
+    @JsonCreator
+    public static ScanMode fromString(final String name) {
+        if (name == null) {
+            throw new IllegalArgumentException("scan_mode must not be null. Valid values are: " + NAMES_MAP.keySet());
+        }
+        final ScanMode mode = NAMES_MAP.get(name.toLowerCase());
+        if (mode == null) {
+            throw new IllegalArgumentException("Invalid scan_mode: " + name + ". Valid values are: " + NAMES_MAP.keySet());
+        }
+        return mode;
+    }
+}

data-prepper-plugins/rds-source/src/main/resources/org/opensearch/dataprepper/transforms/templates/rds-template.yaml

@@ -73,6 +73,7 @@
         folder_partitions:
           depth: "<<FUNCTION_NAME:calculateDepthForRdsSource,PARAMETER:$.<<pipeline-name>>.source.rds.s3_prefix>>"
           max_objects_per_ownership: 50
+          idle_scan_mode: "<<$.<<pipeline-name>>.source.rds.scan_mode>>"
         buckets:
           - bucket:
               name: "<<$.<<pipeline-name>>.source.rds.s3_bucket>>"

data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/configuration/ScanModeTest.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins.source.rds.configuration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ScanModeTest {
+
+    @ParameterizedTest
+    @EnumSource(ScanMode.class)
+    void fromString_returns_correct_value_for_all_enum_constants(final ScanMode mode) {
+        assertThat(ScanMode.fromString(mode.toString()), equalTo(mode));
+    }
+
+    @ParameterizedTest
+    @EnumSource(ScanMode.class)
+    void fromString_is_case_insensitive(final ScanMode mode) {
+        assertThat(ScanMode.fromString(mode.toString().toUpperCase()), equalTo(mode));
+    }
+
+    @ParameterizedTest
+    @EnumSource(ScanMode.class)
+    void toString_returns_non_null(final ScanMode mode) {
+        assertThat(mode.toString(), notNullValue());
+    }
+
+    @Test
+    void fromString_throws_for_invalid_value() {
+        assertThrows(IllegalArgumentException.class, () -> ScanMode.fromString("invalid_mode"));
+    }
+
+    @Test
+    void fromString_throws_for_null() {
+        assertThrows(IllegalArgumentException.class, () -> ScanMode.fromString(null));
+    }
+
+    @Test
+    void idle_optimized_has_correct_string_value() {
+        assertThat(ScanMode.IDLE_OPTIMIZED.toString(), equalTo("idle_optimized"));
+    }
+
+    @Test
+    void continuous_has_correct_string_value() {
+        assertThat(ScanMode.CONTINUOUS.toString(), equalTo("continuous"));
+    }
+}