aws_dynamodb_cdc: add checkpoint_namespace for checkpoint table sharing (#4602)

Add an optional checkpoint_namespace field to the aws_dynamodb_cdc input
so multiple independent pipelines can share a single checkpoint table
without overwriting each other's checkpoints. When set, the namespace is
prefixed to the checkpoint hash key value (StreamArn in default mode,
TableId in global_table mode); shard checkpoints, snapshot progress rows,
and the global-mode resume query are all scoped by it. The default (empty)
leaves checkpoint keys byte-identical to previous behavior, so existing
deployments and checkpoint tables are unaffected. Namespaces containing
'#' are rejected at config validation time.

Author: squiidz

CHANGELOG.md

@@ -24,6 +24,10 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+### Added
+
+- aws_dynamodb_cdc: DynamoDB CDC now supports an optional checkpoint_namespace field, allowing multiple independent pipelines to share a single checkpoint table without overwriting each other's checkpoints. ([@squiidz](https://github.com/squiidz), [#4602](https://github.com/redpanda-data/connect/pull/4602))
+
 ### Fixed
 
 - general: The CGO-enabled distribution binary now embeds the IANA time zone database via the `timetzdata` build tag, matching the other distributions, so `time.LoadLocation` works in minimal runtimes without system tzdata instead of silently falling back to UTC (which shifts JQL date predicates in the `jira` input). ([@squiidz](https://github.com/squiidz), [#4583](https://github.com/redpanda-data/connect/pull/4583))

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

@@ -41,6 +41,7 @@ input:
   aws_dynamodb_cdc:
     tables: []
     checkpoint_table: redpanda_dynamodb_checkpoints
+    checkpoint_namespace: ""
     start_from: trim_horizon
     snapshot_mode: none
 ```
@@ -60,6 +61,7 @@ input:
     table_tag_filter: ""
     table_discovery_interval: 5m
     checkpoint_table: redpanda_dynamodb_checkpoints
+    checkpoint_namespace: ""
     global_table: false
     global_table_replicas: []
     batch_size: 1000
@@ -141,6 +143,8 @@ NOTE: Snapshots use eventually consistent reads and do not provide point-in-time
 
 Checkpoints are stored in a separate DynamoDB table (configured via `checkpoint_table`). This table is created automatically if it does not exist. On restart, the input resumes from the last checkpointed position for each shard. Snapshot progress is also checkpointed, allowing resumption mid-snapshot after failures.
 
+Multiple independent pipelines can share a single checkpoint table by giving each one a distinct `checkpoint_namespace` (for example one namespace per developer or environment). Namespaces isolate checkpoints from each other: a pipeline only sees checkpoints written under its own namespace, so changing (or removing) the namespace causes the pipeline to restart from `start_from`. Note that namespaces do not coordinate consumers — two pipelines sharing the *same* namespace will still overwrite each other's checkpoints.
+
 ### Alternative
 
 For better performance and longer retention (up to 1 year vs 24 hours), consider using Kinesis Data Streams for DynamoDB with the `aws_kinesis` input instead.
@@ -331,6 +335,15 @@ DynamoDB table name for storing checkpoints. Will be created if it doesn't exist
 
 *Default*: `"redpanda_dynamodb_checkpoints"`
 
+=== `checkpoint_namespace`
+
+An optional namespace for checkpoints, allowing multiple independent pipelines (for example one per developer or environment) to share a single checkpoint table without overwriting each other's positions. Checkpoints written under one namespace are invisible to pipelines using a different namespace (or none), so changing this value causes the pipeline to restart from `start_from`. Must not contain `#`.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
 === `global_table`
 
 Provision the checkpoint table as a DynamoDB Global Table (v2) so checkpoints replicate across regions. Requires `global_table_replicas`. When the table is auto-created it is created as a global table; when it already exists, its replicas are reconciled (missing regions are added via `UpdateTable`). The existing table must have been created in global mode (`TableId` hash key) — enabling this against a pre-existing non-global checkpoint table fails fast with a clear error.

internal/impl/aws/dynamodb/checkpoint.go

@@ -48,6 +48,7 @@ type CheckpointerConfig struct {
 	TableName       string // checkpoint table name
 	SourceTable     string // source table name (portable key in global mode)
 	StreamArn       string // current region's stream ARN
+	Namespace       string // optional logical-reader scope, prefixed to the hash key value
 	CheckpointLimit int
 	GlobalTable     bool
 	Region          string   // pipeline's own region (global mode)
@@ -61,6 +62,7 @@ type Checkpointer struct {
 	tableName       string
 	sourceTable     string
 	streamArn       string
+	namespace       string
 	checkpointLimit int
 	globalTable     bool
 	region          string
@@ -87,6 +89,7 @@ func NewCheckpointer(
 		tableName:       cfg.TableName,
 		sourceTable:     cfg.SourceTable,
 		streamArn:       cfg.StreamArn,
+		namespace:       cfg.Namespace,
 		checkpointLimit: cfg.CheckpointLimit,
 		globalTable:     cfg.GlobalTable,
 		region:          cfg.Region,
@@ -149,10 +152,7 @@ func (c *Checkpointer) ensureTableExists(ctx context.Context) error {
 	}
 
 	// Table doesn't exist, create it.
-	hashAttr := checkpointHashKeyDefault
-	if c.globalTable {
-		hashAttr = checkpointHashKeyGlobal
-	}
+	hashAttr := c.hashAttrName()
 	input := &dynamodb.CreateTableInput{
 		AttributeDefinitions: []types.AttributeDefinition{
 			{AttributeName: aws.String(hashAttr), AttributeType: types.ScalarAttributeTypeS},
@@ -246,19 +246,36 @@ func (c *Checkpointer) waitForTableActive(ctx context.Context) error {
 	return nil
 }
 
-// checkpointKey builds the primary key for a shard's checkpoint row. In global
-// mode the hash key is the source table name (portable across regions); in the
-// default mode it is the current stream ARN.
-func (c *Checkpointer) checkpointKey(shardID string) map[string]types.AttributeValue {
+// hashKeyValue returns the value stored in the checkpoint table's hash key:
+// the source table name in global mode (portable across regions), otherwise
+// the current stream ARN. A configured checkpoint namespace is prefixed as
+// "<namespace>#<value>" so independent logical readers sharing one checkpoint
+// table occupy separate partitions.
+func (c *Checkpointer) hashKeyValue() string {
+	v := c.streamArn
 	if c.globalTable {
-		return map[string]types.AttributeValue{
-			checkpointHashKeyGlobal: &types.AttributeValueMemberS{Value: c.sourceTable},
-			checkpointRangeKey:      &types.AttributeValueMemberS{Value: shardID},
-		}
+		v = c.sourceTable
+	}
+	if c.namespace == "" {
+		return v
+	}
+	return c.namespace + "#" + v
+}
+
+// hashAttrName returns the name of the checkpoint table's hash key attribute
+// for the current mode.
+func (c *Checkpointer) hashAttrName() string {
+	if c.globalTable {
+		return checkpointHashKeyGlobal
 	}
+	return checkpointHashKeyDefault
+}
+
+// checkpointKey builds the primary key for a shard's checkpoint row.
+func (c *Checkpointer) checkpointKey(shardID string) map[string]types.AttributeValue {
 	return map[string]types.AttributeValue{
-		checkpointHashKeyDefault: &types.AttributeValueMemberS{Value: c.streamArn},
-		checkpointRangeKey:       &types.AttributeValueMemberS{Value: shardID},
+		c.hashAttrName():   &types.AttributeValueMemberS{Value: c.hashKeyValue()},
+		checkpointRangeKey: &types.AttributeValueMemberS{Value: shardID},
 	}
 }
 
@@ -272,8 +289,8 @@ func (c *Checkpointer) Get(ctx context.Context, shardID string) (string, error)
 		if _, ok := errors.AsType[*types.ResourceNotFoundException](err); ok {
 			return "", nil
 		}
-		return "", fmt.Errorf("getting checkpoint for table=%s stream=%s shard=%s: %w",
-			c.tableName, c.streamArn, shardID, err)
+		return "", fmt.Errorf("getting checkpoint for table=%s key=%s shard=%s: %w",
+			c.tableName, c.hashKeyValue(), shardID, err)
 	}
 
 	if result.Item == nil {
@@ -304,8 +321,8 @@ func (c *Checkpointer) Set(ctx context.Context, shardID, sequenceNumber, approxC
 		Item:      item,
 	})
 	if err != nil {
-		return fmt.Errorf("setting checkpoint for table=%s stream=%s shard=%s seq=%s: %w",
-			c.tableName, c.streamArn, shardID, sequenceNumber, err)
+		return fmt.Errorf("setting checkpoint for table=%s key=%s shard=%s seq=%s: %w",
+			c.tableName, c.hashKeyValue(), shardID, sequenceNumber, err)
 	}
 	return nil
 }
@@ -380,15 +397,15 @@ func (c *Checkpointer) prepareResume(ctx context.Context) error {
 			TableName:              aws.String(c.tableName),
 			KeyConditionExpression: aws.String("TableId = :hash"),
 			ExpressionAttributeValues: map[string]types.AttributeValue{
-				":hash": &types.AttributeValueMemberS{Value: c.sourceTable},
+				":hash": &types.AttributeValueMemberS{Value: c.hashKeyValue()},
 			},
 			ExclusiveStartKey: startKey,
 		})
 		if err != nil {
 			if _, ok := errors.AsType[*types.ResourceNotFoundException](err); ok {
 				return nil
 			}
-			return fmt.Errorf("querying checkpoint partition for %s: %w", c.sourceTable, err)
+			return fmt.Errorf("querying checkpoint partition for %s: %w", c.hashKeyValue(), err)
 		}
 		for _, item := range out.Items {
 			seqAttr, isShardRow := item["SequenceNumber"].(*types.AttributeValueMemberS)
@@ -465,10 +482,7 @@ func (c *Checkpointer) SnapshotProgress(ctx context.Context) (*SnapshotCheckpoin
 		}
 	}
 
-	hashName, hashVal := "StreamArn", c.streamArn
-	if c.globalTable {
-		hashName, hashVal = "TableId", c.sourceTable
-	}
+	hashName, hashVal := c.hashAttrName(), c.hashKeyValue()
 	queryRes, err := c.svc.Query(ctx, &dynamodb.QueryInput{
 		TableName:              aws.String(c.tableName),
 		KeyConditionExpression: aws.String(hashName + " = :hash AND begins_with(ShardID, :snapshot_prefix)"),

internal/impl/aws/dynamodb/checkpoint_test.go

@@ -337,3 +337,190 @@ func TestResolveResume_NonGlobalFallsBackToGet(t *testing.T) {
 	require.Equal(t, resumeExact, d.Mode)
 	require.Equal(t, "seq-42", d.SequenceNumber)
 }
+
+// sharedStoreAPI returns a fakeCheckpointAPI backed by a shared in-memory item
+// store keyed by "<hashValue>|<shardID>", so multiple Checkpointers can
+// exercise cross-namespace visibility against one "table".
+func sharedStoreAPI(store map[string]map[string]types.AttributeValue) *fakeCheckpointAPI {
+	keyOf := func(m map[string]types.AttributeValue) string {
+		attrStr := func(name string) string {
+			if v, ok := m[name].(*types.AttributeValueMemberS); ok {
+				return v.Value
+			}
+			return ""
+		}
+		// Global-mode items carry a non-key StreamArn attribute alongside the
+		// TableId hash key, so prefer TableId when present.
+		hash := attrStr(checkpointHashKeyGlobal)
+		if hash == "" {
+			hash = attrStr(checkpointHashKeyDefault)
+		}
+		return hash + "|" + attrStr(checkpointRangeKey)
+	}
+	return &fakeCheckpointAPI{
+		describeTable: func(context.Context, *dynamodb.DescribeTableInput, ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) {
+			return &dynamodb.DescribeTableOutput{Table: &types.TableDescription{TableStatus: types.TableStatusActive}}, nil
+		},
+		putItem: func(_ context.Context, in *dynamodb.PutItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error) {
+			store[keyOf(in.Item)] = in.Item
+			return &dynamodb.PutItemOutput{}, nil
+		},
+		getItem: func(_ context.Context, in *dynamodb.GetItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error) {
+			return &dynamodb.GetItemOutput{Item: store[keyOf(in.Key)]}, nil
+		},
+	}
+}
+
+func TestCheckpointer_NamespacePrefixesHashKeyValue(t *testing.T) {
+	cases := []struct {
+		name      string
+		namespace string
+		global    bool
+		wantAttr  string
+		wantValue string
+	}{
+		{"default mode no namespace", "", false, "StreamArn", "arn:A"},
+		{"default mode namespaced", "dev-alice", false, "StreamArn", "dev-alice#arn:A"},
+		{"global mode no namespace", "", true, "TableId", "t"},
+		{"global mode namespaced", "dev-alice", true, "TableId", "dev-alice#t"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var put *dynamodb.PutItemInput
+			api := &fakeCheckpointAPI{
+				describeTable: func(context.Context, *dynamodb.DescribeTableInput, ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) {
+					desc := &types.TableDescription{TableStatus: types.TableStatusActive}
+					if tc.global {
+						desc.KeySchema = globalTableKeySchema()
+						desc.Replicas = []types.ReplicaDescription{
+							{RegionName: aws.String("us-east-1")},
+							{RegionName: aws.String("us-west-2")},
+						}
+					}
+					return &dynamodb.DescribeTableOutput{Table: desc}, nil
+				},
+				putItem: func(_ context.Context, in *dynamodb.PutItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error) {
+					put = in
+					return &dynamodb.PutItemOutput{}, nil
+				},
+			}
+			cfg := CheckpointerConfig{
+				TableName: "cps", SourceTable: "t", StreamArn: "arn:A",
+				Namespace: tc.namespace, CheckpointLimit: 1,
+			}
+			if tc.global {
+				cfg.GlobalTable = true
+				cfg.Region = "us-east-1"
+				cfg.ReplicaRegions = []string{"us-west-2"}
+			}
+			c, err := NewCheckpointer(context.Background(), api, cfg, checkpointTestLogger())
+			require.NoError(t, err)
+
+			require.NoError(t, c.Set(context.Background(), "shard-1", "seq-1", ""))
+			require.Equal(t, tc.wantValue, put.Item[tc.wantAttr].(*types.AttributeValueMemberS).Value)
+			require.Equal(t, "shard-1", put.Item[checkpointRangeKey].(*types.AttributeValueMemberS).Value,
+				"range key must never be namespaced")
+		})
+	}
+}
+
+func TestCheckpointer_NamespaceIsolation(t *testing.T) {
+	store := map[string]map[string]types.AttributeValue{}
+	api := sharedStoreAPI(store)
+
+	newCP := func(ns string) *Checkpointer {
+		c, err := NewCheckpointer(context.Background(), api, CheckpointerConfig{
+			TableName: "cps", SourceTable: "t", StreamArn: "arn:A",
+			Namespace: ns, CheckpointLimit: 1,
+		}, checkpointTestLogger())
+		require.NoError(t, err)
+		return c
+	}
+	alice, bob, unscoped := newCP("dev-alice"), newCP("dev-bob"), newCP("")
+
+	require.NoError(t, alice.Set(context.Background(), "shard-1", "seq-alice", ""))
+	require.NoError(t, unscoped.Set(context.Background(), "shard-1", "seq-legacy", ""))
+
+	got, err := alice.Get(context.Background(), "shard-1")
+	require.NoError(t, err)
+	require.Equal(t, "seq-alice", got, "a namespace must read back its own checkpoint")
+
+	got, err = bob.Get(context.Background(), "shard-1")
+	require.NoError(t, err)
+	require.Empty(t, got, "a namespace must not see another namespace's checkpoint")
+
+	got, err = unscoped.Get(context.Background(), "shard-1")
+	require.NoError(t, err)
+	require.Equal(t, "seq-legacy", got, "un-namespaced reader must not see namespaced rows")
+}
+
+func TestSnapshotCheckpoints_UseNamespacedHashValue(t *testing.T) {
+	var getIn *dynamodb.GetItemInput
+	var queryIn *dynamodb.QueryInput
+	var putIn *dynamodb.PutItemInput
+	api := &fakeCheckpointAPI{
+		describeTable: func(context.Context, *dynamodb.DescribeTableInput, ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) {
+			return &dynamodb.DescribeTableOutput{Table: &types.TableDescription{TableStatus: types.TableStatusActive}}, nil
+		},
+		getItem: func(_ context.Context, in *dynamodb.GetItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error) {
+			getIn = in
+			return &dynamodb.GetItemOutput{}, nil
+		},
+		query: func(_ context.Context, in *dynamodb.QueryInput, _ ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error) {
+			queryIn = in
+			return &dynamodb.QueryOutput{}, nil
+		},
+		putItem: func(_ context.Context, in *dynamodb.PutItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error) {
+			putIn = in
+			return &dynamodb.PutItemOutput{}, nil
+		},
+	}
+	c, err := NewCheckpointer(context.Background(), api, CheckpointerConfig{
+		TableName: "cps", SourceTable: "t", StreamArn: "arn:A",
+		Namespace: "dev-alice", CheckpointLimit: 1,
+	}, checkpointTestLogger())
+	require.NoError(t, err)
+
+	_, err = c.SnapshotProgress(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, "dev-alice#arn:A", getIn.Key["StreamArn"].(*types.AttributeValueMemberS).Value,
+		"snapshot completion sentinel must use the namespaced hash value")
+	require.Equal(t, "dev-alice#arn:A", queryIn.ExpressionAttributeValues[":hash"].(*types.AttributeValueMemberS).Value,
+		"snapshot segment query must use the namespaced hash value")
+
+	require.NoError(t, c.UpdateSnapshotProgress(context.Background(), 3, nil, 42))
+	require.Equal(t, "dev-alice#arn:A", putIn.Item["StreamArn"].(*types.AttributeValueMemberS).Value,
+		"snapshot segment writes must use the namespaced hash value")
+	require.Equal(t, "snapshot#segment#3", putIn.Item[checkpointRangeKey].(*types.AttributeValueMemberS).Value)
+}
+
+func TestPrepareResume_UsesNamespacedHashValue(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{
+				TableName:   aws.String("cps"),
+				TableStatus: types.TableStatusActive,
+				KeySchema:   globalTableKeySchema(),
+				Replicas: []types.ReplicaDescription{
+					{RegionName: aws.String("us-east-1")},
+					{RegionName: aws.String("us-west-2")},
+				},
+			}}, 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", CheckpointLimit: 1,
+		Namespace: "dev-alice", GlobalTable: true, Region: "us-east-1", ReplicaRegions: []string{"us-west-2"},
+	}, checkpointTestLogger())
+	require.NoError(t, err)
+
+	_, err = c.ResolveResume(context.Background(), "shard-1")
+	require.NoError(t, err)
+	require.Equal(t, "dev-alice#t", queryIn.ExpressionAttributeValues[":hash"].(*types.AttributeValueMemberS).Value,
+		"global-mode resume partition query must use the namespaced hash value")
+}