connect-cdc-dynamodb

# Redpanda Connect CDC: AWS DynamoDB The `aws_dynamodb_cdc` input reads change events from DynamoDB Streams into Redpanda or any Kafka-compatible broker. It manages shard lifecycle automatically, writes checkpoints to a dedicated DynamoDB table, and supports an optional initial Scan snapshot before switching to streaming. Available from Redpanda Connect v4.79.0 with status Stable. It is classified as an **enterprise**-tier connector, and its source is an Enterprise-licensed file under the Redpanda Community License. Unlike the other enterprise `*_cdc` inputs, however, the connector's source currently enforces **no runtime Enterprise license gate**. Three table-discovery modes cover single-table, explicit-list, and tag-based multi-table scenarios. The 24-hour DynamoDB Streams retention window is the key operational constraint: the connector must remain running or resume within that window or it re-runs a snapshot to avoid data loss. ## Quickstart ### 1. Enable DynamoDB Streams on your table ```bash # Enable NEW_AND_OLD_IMAGES stream on an existing table aws dynamodb update-table \ --table-name orders \ --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES # Verify aws dynamodb describe-table --table-name orders \ --query "Table.StreamSpecification" ``` ### 2. Create the IAM policy ```bash # Save as dynamodb-cdc-policy.json (replace REGION, ACCOUNT_ID, TABLE_NAME) cat > dynamodb-cdc-policy.json <<'EOF' { "Version": "2012-10-17", "Statement": [ { "Sid": "DynamoDBStreams", "Effect": "Allow", "Action": [ "dynamodb:DescribeStream", "dynamodb:GetRecords", "dynamodb:GetShardIterator" ], "Resource": "arn:aws:dynamodb:REGION:ACCOUNT_ID:table/orders/stream/*" }, { "Sid": "DescribeSourceTable", "Effect": "Allow", "Action": ["dynamodb:DescribeTable"], "Resource": "arn:aws:dynamodb:REGION:ACCOUNT_ID:table/orders" }, { "Sid": "SnapshotScan", "Effect": "Allow", "Action": ["dynamodb:Scan"], "Resource": "arn:aws:dynamodb:REGION:ACCOUNT_ID:table/orders" }, { "Sid": "CheckpointTable", "Effect": "Allow", "Action": [ "dynamodb:CreateTable", "dynamodb:DescribeTable", "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:Query" ], "Resource": "arn:aws:dynamodb:REGION:ACCOUNT_ID:table/redpanda_dynamodb_checkpoints" }, { "Sid": "TagDiscovery", "Effect": "Allow", "Action": ["dynamodb:ListTables", "dynamodb:ListTagsOfResource"], "Resource": "*" } ] } EOF aws iam create-policy \ --policy-name redpanda-dynamodb-cdc \ --policy-document file://dynamodb-cdc-policy.json ``` ### 3. Create the pipeline config ```yaml # pipeline.yaml — snapshot existing data then stream changes input: aws_dynamodb_cdc: tables: [orders] start_from: trim_horizon snapshot_mode: snapshot_and_cdc snapshot_segments: 4 snapshot_throttle: 100ms snapshot_deduplicate: true checkpoint_table: redpanda_dynamodb_checkpoints region: us-east-1 output: redpanda: seed_brokers: ["localhost:9092"] topic: '${! meta("dynamodb_table") }-cdc' ``` ### 4. Run it ```bash # With environment credentials (IAM role / ~/.aws/credentials): rpk connect run pipeline.yaml # With explicit static credentials: rpk connect run pipeline.yaml \ -s "input.aws_dynamodb_cdc.credentials.id=${AWS_ACCESS_KEY_ID}" \ -s "input.aws_dynamodb_cdc.credentials.secret=${AWS_SECRET_ACCESS_KEY}" ``` ### 5. Inspect messages Each message body is a JSON object with this shape. For an INSERT event the `oldImage` key is absent (not null); for a REMOVE event the `newImage` key is absent. The keys `keys`, `newImage`, `oldImage`, and `sizeBytes` are only present when the corresponding field is non-nil in the stream record: ```json { "tableName": "orders", "eventID": "abc123", "eventName": "INSERT", "eventVersion": "1.1", "eventSource": "aws:dynamodb", "awsRegion": "us-east-1", "dynamodb": { "sequenceNumber": "000000000000000000001", "streamViewType": "NEW_AND_OLD_IMAGES", "keys": { "orderId": "ORD-001" }, "newImage": { "orderId": "ORD-001", "status": "pending", "total": 99.99 }, "sizeBytes": 128 } } ``` Snapshot records use `"eventName": "READ"` and include only `newImage`. Metadata fields differ between CDC and snapshot records: | Metadata key | CDC records | Snapshot records | |---|---|---| | `dynamodb_event_name` | `INSERT`, `MODIFY`, or `REMOVE` | `READ` | | `dynamodb_table` | table name | table name | | `dynamodb_shard_id` | shard ID string | _not present_ | | `dynamodb_sequence_number` | stream sequence number | _not present_ | | `dynamodb_snapshot_segment` | _not present_ | segment index (e.g. `"0"`) | Note: `dynamodb_shard_id` and `dynamodb_sequence_number` are absent on snapshot records — not empty strings. Checking `meta("dynamodb_shard_id")` on a snapshot record returns a missing-key error in Bloblang. ## Snapshot Modes | Mode | Behavior | |---|---| | `none` (default) | CDC streaming only, from existing stream position | | `snapshot_only` | One-time full Scan, then stops | | `snapshot_and_cdc` | Full Scan first, then streams ongoing changes | `snapshot_and_cdc` starts CDC shard readers _before_ the Scan begins so no writes during the snapshot window are missed. Use `snapshot_deduplicate: true` (default) to suppress records that appear in both. ## Multi-Table Discovery ```yaml # By explicit list input: aws_dynamodb_cdc: table_discovery_mode: includelist tables: [orders, customers, products] region: us-east-1 --- # By tag — auto-discover all tables tagged stream-enabled:true input: aws_dynamodb_cdc: table_discovery_mode: tag table_tag_filter: "stream-enabled:true" table_discovery_interval: 5m region: us-east-1 --- # Multi-criteria tag filter (AND across keys, OR within a key) input: aws_dynamodb_cdc: table_discovery_mode: tag table_tag_filter: "environment:prod,staging;team:data,analytics" region: us-east-1 ``` Snapshot modes are supported only when the effective configuration is single-table at config-validation time: `table_discovery_mode: single`, or `table_discovery_mode: includelist` with exactly one table. `tag` discovery mode always rejects `snapshot_mode` other than `none` at startup, even if the tag filter would match only one table — it is treated as multi-table regardless. ## Per-Table Routing Use a Bloblang mapping or `switch` output to route each table's events to a dedicated topic: ```yaml output: switch: cases: - check: 'meta("dynamodb_table") == "orders"' output: redpanda: seed_brokers: ["localhost:9092"] topic: orders-cdc - check: 'meta("dynamodb_table") == "customers"' output: redpanda: seed_brokers: ["localhost:9092"] topic: customers-cdc ``` ## Operational Notes - **Stream retention:** DynamoDB Streams retain records for 24 hours. If the connector is down longer than that, it re-runs a snapshot on next start (in `snapshot_and_cdc` mode). - **Checkpoint table:** Created automatically with pay-per-request billing. Uses `(StreamArn, ShardID)` as the primary key. Give each pipeline sharing one table a distinct `checkpoint_namespace`; namespaces isolate positions but do not coordinate consumers, so two pipelines in the *same* namespace still overwrite each other. - **`start_from` applies to a fresh pipeline only:** `latest` is honoured on the first shard discovery when no checkpoint state exists under the namespace. Once state exists, later-discovered and checkpoint-less shards (rotation children, shards seen after a restart) always start at `trim_horizon` so their backlog is never skipped — at-least-once wins over the configured start position. - **Delivery on rejection (`auto_replay_nacks`, new in 4.106.0, default `true`):** rejected messages are retried in-process, so a transient downstream failure needs no restart to recover. Setting it to `false` **deletes** rejected messages — an explicit opt-in to dropping data, not a tuning knob. Snapshot checkpoints are persisted only after a batch and everything before it is acked, so under the default a rejected snapshot batch is redelivered rather than skipped. - **Kinesis alternative:** For up to 1-year retention, enable Kinesis Data Streams for DynamoDB and use the `aws_kinesis` input instead. - **License and tier:** `aws_dynamodb_cdc` is an **enterprise**-tier connector whose source is an Enterprise-licensed file under the Redpanda Community License. Unlike the other enterprise `*_cdc` inputs, its source currently enforces no runtime Enterprise license gate (it is registered as Stable) — but treat it as enterprise-tier, since that classification (not the currently-absent gate) is authoritative. ## Enterprise Features on the Destination Topic The `aws_dynamodb_cdc` input is an enterprise-tier connector (its source currently enforces no runtime license gate). Separately, the Redpanda topic and cluster the CDC events land in can use Redpanda Enterprise differentiators (each requires a valid Redpanda Enterprise license on the destination cluster): - **Iceberg Topics** — land CDC events directly into an Apache Iceberg table for analytics. Per-topic: `redpanda.iceberg.mode` (`key_value` | `value_schema_id_prefix` | `value_schema_latest` | `disabled`), `redpanda.iceberg.delete`, `redpanda.iceberg.partition.spec`, `redpanda.iceberg.target.lag.ms`, `redpanda.iceberg.invalid.record.action` (`drop` | `dlq_table`); cluster: `iceberg_enabled`. - **Tiered Storage** — `redpanda.remote.write` + `redpanda.remote.read` (cluster `cloud_storage_enabled`) extend CDC retention far beyond DynamoDB Streams' 24h window. - **Remote Read Replicas** — `redpanda.remote.readreplica` for read-only, object-storage-served copies of the CDC topic in a remote cluster. - **Shadow Linking / Shadowing** — offset-preserving cross-cluster DR for the CDC topic, managed with the `rpk shadow` family (`create`/`list`/`describe`/`status`/`update`/`failover`/`delete`/`config-generate`). Failover preserves consumer offsets on the DR cluster. - **Server-side Schema ID Validation** — `redpanda.value.schema.id.validation` + `redpanda.value.subject.name.strategy` (cluster `enable_schema_id_validation` = `none`/`redpanda`/`compat`). - **RBAC, Audit Logging, OIDC/Kerberos, FIPS** — secure the CDC topic and pipeline. - **Redpanda Connect Enterprise** (separate Connect license) — secrets management, configuration service, allow/deny lists. See [Enterprise Redpanda Features](references/enterprise-redpanda-features.md) for grounded nested config keys, defaults, license-expiration behavior, and `rpk` commands. ## Reference Directory - [Config Reference](references/config-reference.md): Every `aws_dynamodb_cdc` config field with type, default, and description, grounded in source. - [DynamoDB Setup](references/setup-dynamodb.md): Enabling DynamoDB Streams, stream view types, checkpoint table schema, IAM policy, and stream retention. - [Pipeline and Output](references/pipeline-and-output.md): Full runnable pipelines, message/metadata shape, snapshot vs CDC modes, deduplication, and restart behavior. - [Enterprise Redpanda Features](references/enterprise-redpanda-features.md): Iceberg Topics, Tiered Storage, Remote Read Replicas, Shadow Linking (cross-cluster DR), server-side Schema ID Validation, RBAC/Audit Logging/OIDC/Kerberos/FIPS, and Redpanda Connect Enterprise — nested config keys, defaults, and which license gates each.