aws_dynamodb_cdc: ack-gate snapshot checkpoints and fix shard start_from handling (CON-504)

docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc

@@ -43,6 +43,7 @@ input:
     checkpoint_table: redpanda_dynamodb_checkpoints
     checkpoint_namespace: ""
     start_from: trim_horizon
+    auto_replay_nacks: true
     snapshot_mode: none
 ```
 
@@ -69,6 +70,7 @@ input:
     start_from: trim_horizon
     checkpoint_limit: 1000
     max_tracked_shards: 10000
+    auto_replay_nacks: true
     throttle_backoff: 100ms
     snapshot_mode: none
     snapshot_segments: 1
@@ -382,7 +384,9 @@ Time to wait between polling attempts when no records are available.
 
 === `start_from`
 
-Where to start reading when no checkpoint exists. `trim_horizon` starts from the oldest available record, `latest` starts from new records.
+Where to start reading on a genuinely fresh pipeline (no checkpoint state exists yet under this `checkpoint_namespace` for the stream). `trim_horizon` starts from the oldest available record, `latest` starts from new records.
+
+`latest` is honoured only on that first discovery: once any checkpoint state exists, shards discovered later - rotation children found by the periodic refresh, and any checkpoint-less shard after a restart - always start at `trim_horizon` so their backlog is never skipped. In practice a restart under `latest` therefore replays from each shard's oldest retained record rather than only new records; at-least-once delivery takes precedence over the configured start position.
 
 
 *Type*: `string`
@@ -412,6 +416,15 @@ Maximum number of shards to track simultaneously. Prevents memory issues with ex
 
 *Default*: `10000`
 
+=== `auto_replay_nacks`
+
+Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.
+
+
+*Type*: `bool`
+
+*Default*: `true`
+
 === `throttle_backoff`
 
 Time to wait when applying backpressure due to too many in-flight messages.

internal/impl/aws/dynamodb/checkpoint.go

@@ -332,6 +332,38 @@ func (c *Checkpointer) CheckpointLimit() int {
 	return c.checkpointLimit
 }
 
+// HasAnyState reports whether any checkpoint state (shard checkpoints or
+// snapshot progress) exists under this pipeline's namespace and stream/table
+// key. Used to scope start_from to genuinely fresh pipelines: shards that
+// appear once state exists are stream-rotation children whose backlog must
+// not be skipped.
+func (c *Checkpointer) HasAnyState(ctx context.Context) (bool, error) {
+	result, err := c.svc.Query(ctx, &dynamodb.QueryInput{
+		TableName:              aws.String(c.tableName),
+		KeyConditionExpression: aws.String("#hk = :hv"),
+		ExpressionAttributeNames: map[string]string{
+			"#hk": c.hashAttrName(),
+		},
+		ExpressionAttributeValues: map[string]types.AttributeValue{
+			":hv": &types.AttributeValueMemberS{Value: c.hashKeyValue()},
+		},
+		Limit: aws.Int32(1),
+		// This probe is the sole gate for honoring start_from: latest. An
+		// eventually consistent read could miss checkpoint rows written
+		// moments before a crash-restart and reposition shards at LATEST,
+		// silently skipping their backlog; Limit 1 makes the strong read
+		// nearly free.
+		ConsistentRead: aws.Bool(true),
+	})
+	if err != nil {
+		if _, ok := errors.AsType[*types.ResourceNotFoundException](err); ok {
+			return false, nil
+		}
+		return false, fmt.Errorf("probing checkpoint state for table=%s key=%s: %w", c.tableName, c.hashKeyValue(), err)
+	}
+	return len(result.Items) > 0, nil
+}
+
 type resumeMode int
 
 const (

internal/impl/aws/dynamodb/checkpoint_test.go

@@ -277,6 +277,49 @@ func shardRow(streamArn, shardID, seq, ts string) map[string]types.AttributeValu
 	return row
 }
 
+func TestHasAnyState(t *testing.T) {
+	t.Run("false on an empty partition", func(t *testing.T) {
+		c := globalCheckpointerWithPartition(t, "arn:A", nil)
+		has, err := c.HasAnyState(context.Background())
+		require.NoError(t, err)
+		require.False(t, has)
+	})
+
+	t.Run("true when any row exists", func(t *testing.T) {
+		c := globalCheckpointerWithPartition(t, "arn:A", []map[string]types.AttributeValue{
+			shardRow("arn:A", "shard-1", "seq-9", ""),
+		})
+		has, err := c.HasAnyState(context.Background())
+		require.NoError(t, err)
+		require.True(t, has)
+	})
+
+	t.Run("queries the namespaced hash key", func(t *testing.T) {
+		var queryIn *dynamodb.QueryInput
+		api := &fakeCheckpointAPI{
+			describeTable: func(context.Context, *dynamodb.DescribeTableInput, ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) {
+				return &dynamodb.DescribeTableOutput{Table: &types.TableDescription{TableStatus: types.TableStatusActive}}, nil
+			},
+			query: func(_ context.Context, in *dynamodb.QueryInput, _ ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error) {
+				queryIn = in
+				return &dynamodb.QueryOutput{}, nil
+			},
+		}
+		c, err := NewCheckpointer(context.Background(), api, CheckpointerConfig{
+			TableName: "cps", SourceTable: "t", StreamArn: "arn:A",
+			Namespace: "dev", CheckpointLimit: 1, Region: "us-east-1",
+		}, checkpointTestLogger())
+		require.NoError(t, err)
+
+		_, err = c.HasAnyState(context.Background())
+		require.NoError(t, err)
+		require.NotNil(t, queryIn)
+		hv, ok := queryIn.ExpressionAttributeValues[":hv"].(*types.AttributeValueMemberS)
+		require.True(t, ok)
+		require.Equal(t, "dev#arn:A", hv.Value)
+	})
+}
+
 func TestCDCCheckpointProbeNeeded(t *testing.T) {
 	// Only an exact, same-region resume can be a stale checkpoint: its sequence
 	// number belongs to this stream, so a failed iterator means trimmed data.