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
@@ -412,6 +414,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,32 @@ 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),
+	})
+	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.