Add disable_checkpointing flag for ddb source stream (#5981)

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

Author: graytaylor0

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/configuration/StreamConfig.java

@@ -16,6 +16,9 @@ public class StreamConfig {
     @JsonProperty("view_on_remove")
     private StreamViewType viewForRemoves = StreamViewType.NEW_IMAGE;
 
+    @JsonProperty("disable_checkpointing")
+    private boolean disableCheckpointing = false;
+
     public StreamStartPosition getStartPosition() {
         return startPosition;
     }
@@ -24,4 +27,6 @@ public StreamViewType getStreamViewForRemoves() {
         return viewForRemoves;
     }
 
+    public boolean isDisableCheckpointing() { return disableCheckpointing; }
+
 }

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumerFactory.java

@@ -78,6 +78,9 @@ public Runnable createConsumer(final StreamPartition streamPartition,
         if (progressState.isPresent()) {
             // We can't checkpoint with acks yet
             sequenceNumber = acknowledgementSet == null ? null : progressState.get().getSequenceNumber();
+            if (streamConfig.isDisableCheckpointing()) {
+                sequenceNumber = null;
+            }
             waitForExport = progressState.get().shouldWaitForExport();
             if (progressState.get().getStartTime() != 0) {
                 startTime = Instant.ofEpochMilli(progressState.get().getStartTime());

data-prepper-plugins/dynamodb-source/src/test/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumerFactoryTest.java

@@ -9,9 +9,11 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
 import org.opensearch.dataprepper.model.buffer.Buffer;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.record.Record;
@@ -26,13 +28,15 @@
 import software.amazon.awssdk.services.dynamodb.model.GetShardIteratorRequest;
 import software.amazon.awssdk.services.dynamodb.model.GetShardIteratorResponse;
 import software.amazon.awssdk.services.dynamodb.model.InternalServerErrorException;
+import software.amazon.awssdk.services.dynamodb.model.ShardIteratorType;
 import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
 
 import java.time.Instant;
 import java.util.Optional;
 import java.util.UUID;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.ArgumentMatchers.any;
@@ -123,6 +127,31 @@ public void test_create_shardConsumer_correctly() {
         verify(streamApiInvocations).increment();
     }
 
+    @Test
+    public void test_create_shardConsumer_correctly_with_is_disable_checkpointing_enabled_starts_from_trim_horizon() {
+
+        final AcknowledgementSet acknowledgementSet = mock(AcknowledgementSet.class);
+        when(streamConfig.isDisableCheckpointing()).thenReturn(true);
+        StreamProgressState state = new StreamProgressState();
+        state.setWaitForExport(false);
+        state.setSequenceNumber(UUID.randomUUID().toString());
+        state.setStartTime(Instant.now().toEpochMilli());
+        streamPartition = new StreamPartition(streamArn, shardId, Optional.of(state));
+
+        ShardConsumerFactory consumerFactory = new ShardConsumerFactory(coordinator, dynamoDbStreamsClient, pluginMetrics, dynamoDBSourceAggregateMetrics, buffer, streamConfig);
+        Runnable consumer = consumerFactory.createConsumer(streamPartition, acknowledgementSet, null);
+        assertThat(consumer, notNullValue());
+
+        final ArgumentCaptor<GetShardIteratorRequest> captor = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
+        verify(dynamoDbStreamsClient).getShardIterator(captor.capture());
+
+        final GetShardIteratorRequest getShardIteratorRequest = captor.getValue();
+        assertThat(getShardIteratorRequest.sequenceNumber(), equalTo(null));
+        assertThat(getShardIteratorRequest.shardIteratorType(), equalTo(ShardIteratorType.TRIM_HORIZON));
+
+        verify(streamApiInvocations).increment();
+    }
+
     @Test
     public void test_create_shardConsumer_for_closedShards() {
         // For ending sequence number != null