# Amazon DynamoDB CDC Source Connector for Confluent Cloud
The fully managed Amazon DynamoDB CDC Source connector for Confluent Cloud moves
data from Amazon DynamoDB to a Apache Kafka® topic.
#### NOTE
If you require private networking for fully managed connectors, make sure to set up the proper
networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud).
## Features
The Amazon DynamoDB CDC Source connector includes the following features:
* **At least once delivery**: The connector guarantees that records are delivered at least once.
* **IAM user authentication**: The connector supports authenticating to DynamoDB
using IAM user access credentials.
* **Provider integration support**: The connector supports IAM role-based authorization using Confluent Provider
Integration. For more information about provider integration setup, see the [IAM roles authentication](#cc-amazon-dynamodb-cdc-source-setup-connection).
* **Customizable API endpoints**: The connector allows you to specify an AWS
DynamoDB API and Resource Group Tag API endpoint.
* **Kafka cluster authentication customization**: The connector supports
authenticating a Kafka cluster using API keys and/or service accounts.
* **Snapshot mode customization**: The connector allows you to configure either
of the following modes for snapshots:
- **SNAPSHOT**: Only allows a one-time scan (Snapshot) of the existing data in
the source tables simultaneously.
- **CDC**: Only allows CDC with DynamoDB streams without initial snapshot
for all streams simultaneously.
- **SNAPSHOT_CDC** (Default): Allows an initial snapshot of all configured
tables and once the snapshot is complete, starts CDC streaming using
DynamoDB streams.
#### NOTE
In `SNAPSHOT_CDC` mode, the connector requires at least one CDC event to
fully complete the snapshot phase and transition to CDC. If a DynamoDB table
has no write activity after the snapshot completes, the connector will remain
in a transitional state. To complete the transition, perform a write operation
on the table, or wait for natural write activity.
* **Seamless table streaming**: The connector support the following two modes to
provide seamless table streaming:
- **TAG_MODE**: Auto-discover multiple DynamoDB tables and stream
simultaneously (that is, `dynamodb.table.discovery.mode` is set to `TAG`
- **INCLUDELIST_MODE**: Explicitly specify/select specific multiple DynamoDB
table names and stream simultaneously (that is,
`dynamodb.table.discovery.mode` is set to `INCLUDELIST`).
* **Automatic topic creation**: The connector supports the auto-creation of
topics with the name of the table, with a customer-provided prefix and suffix
using [TopicRegexRouter SMT
(SMT)](/platform/current/connect/transforms/topicregexrouter.html).
* **Supported data formats**: The connector supports Avro, Protobuf, and JSON
Schema output formats. [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be
enabled to use a Schema Registry-based format (for example, Avro, JSON Schema, or
Protobuf).
* **Schema management**: The connector supports Schema Registry, Schema Context
and Reference Subject Naming Strategy.
The connector can infer the schema of nested fields and validate records against
the inferred schema. To configure this functionality, set the `document.content.format`
configuration property to `schema`. Depending on your requirements, you may also
need to configure `document.schema.source` configuration property. For more
information, see [Schema validation](https://docs.confluent.io/cloud/current/connectors/cc-amazon-dynamodb-cdc-source.html#schema-validation).
* **AWS DynamoDB scanning capabilities**: The connector includes the following
scanning capabilities:
- **Parallel Scans**: A DynamoDB table can be logically divided into multiple
segments. The connector will divide the table into five logical segments and
scan these segments in parallel.
- **Pagination**: Tables are scanned sequentially, and a scan result’s
response can fetch no more than 1 MB of data. Since tables can be large, scan
request responses are paginated. With each response, a `LastEvaluatedKey` is
returned. This `LastEvaluatedKey` from a scan response should be used as the
`ExclusiveStartKey` for the next scan request. If no `LastEvaluatedKey` is
returned, it indicates that the end of the result set has been reached.
- **Non-isolated scans**: To scan an entire table, the task continues making
multiple subsequent scan requests by submitting the appropriate
`exclusiveStartKey` with each request. For a large table, this process may
take hours, and the snapshot will capture items as they are at the time a
scan request is made, and not from when the snapshot operation was started.
- **Eventual Consistency**: By default, a scan uses eventually consistent
reads when accessing items in a table. Therefore, the results from a
consistent scan may not reflect the latest item changes at the
time the scan iterates through each item in the table. A snapshot, on the
other hand, only captures data from committed transactions. As a result, a
scan will not include data from ongoing uncommitted transactions.
Additionally, a snapshot does not need to manage or track any ongoing
transactions on a DynamoDB table.
* **Custom offset support**: The connector allows you to configure [custom
offsets](../offsets.md#connect-custom-offsets) using the Confluent Cloud user interface to prevent
data loss and data duplication.
* **Tombstone event and record deletion management**: The connector allows you
to manage tombstone events and deleted records. Note that when the connector
detects a delete event, it creates two event messages:
- A delete event message with `op` type `d` and `document` field with
the table primary key:
```json
{
"op": "d"
"key": {
"id": "5028"
},
"value": {
"document": "5028"
},
}
```
- A tombstone record with Kafka Record Key as the table primary key value and
Kafka Record Value as `null`:
```json
{
"key": {
"id": "5028"
},
"value": null
}
```
Note that Kafka log compaction uses this to know that it can delete all
messages for this key.
**Tombstone message sample**
```json
{
"topic": "table1",
"key": {
"id": "5028"
},
"value": null,
"partition": 0,
"offset": 1
}
```
* **Lease table prefix customization**: The connector supports naming lease
tables with a prefix.
* **Secret manager integration**: The connector supports secret manager integration. For `Access Keys` based
authentication, the connector can retrieve the following configurations from an integrated secret manager
at runtime as needed.
| **Secret manager managed configuration** | **Type** |
|--------------------------------------------|------------|
| `aws.access.key.id` | `PASSWORD` |
| `aws.secret.access.key` | `PASSWORD` |
For more information, see [Create a secret manager integration in Confluent Cloud](../secret-manager-integration/overview.md#cloud-secret-manager-quickstart).
## Limitations
Be sure to review the following information.
* For connector limitations, see [Amazon DynamoDB CDC Source Connector](../limits.md#cc-amazon-dynamodb-cdc-source-limits)
limitations.
* If you plan to use one or more Single Message Transformations (SMTs), see
[SMT Limitations](../single-message-transforms.md#cc-single-message-transforms-limitations).
## IAM policy DynamoDB CDC
The following permissions are required for the Amazon DynamoDB CDC Source
connector:
* DescribeTable
* DescribeStream
* ListTagsOfResource
* DescribeLimits
* GetRecords
* GetShardIterator
* Scan
* ListStreams
* ListTables
* ListGlobalTables
* GetResources
You can copy the following JSON policy. For more information, see [Creating
policies on the JSON tab](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_create-console.html#access_policies_create-json-editor).
```json
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"dynamodb:DescribeTable",
"dynamodb:DescribeStream",
"dynamodb:ListTagsOfResource",
"dynamodb:DescribeLimits",
"dynamodb:GetRecords",
"dynamodb:GetShardIterator",
"dynamodb:Scan"
],
"Resource": [
"arn:aws:dynamodb:*:*:table/*"
]
},
{
"Effect": "Allow",
"Action": [
"dynamodb:CreateTable",
"dynamodb:DescribeTable",
"dynamodb:GetItem",
"dynamodb:PutItem",
"dynamodb:Scan",
"dynamodb:UpdateItem",
"dynamodb:DeleteItem"
],
"Resource": [
"arn:aws:dynamodb:*:*:table/-*" // regex for checkpointing tables
]
},
{
"Effect": "Allow",
"Action": [
"dynamodb:ListStreams",
"dynamodb:ListTables",
"dynamodb:ListGlobalTables",
"tag:GetResources"
],
"Resource": [
"*"
]
}
]
}
```
Note the following with the previous JSON policy:
Destination DynamoDB tables as data sources
: - The first highlighted section contains the permissions required for the DynamoDB Table from which data is sourced.
Target Checkpointing table created by the connector
: - The second highlighted section contains the permissions required for the Checkpointing table created during the CDC phase.
- `` configures the prefix for the checkpointing table created.
Destination DynamoDB streams as data sources
: - The third highlighted section contains the permissions required for DynamoDB Streams from which data is sourced.
## Offset guidance
The connector currently works with the following offsets:
```json
[
{
// TypeA: this offset is logged once per table and should not be modified
"partition": {
"tableName": "",
"cdcStarted": "true"
},
"offset": {
"cdcStarted": "true"
}
},
{
// TypeB: This offset is logged 5 times (segment-0, ... segment-4) for each table and should not be modified.
"partition": {
"segment": "0",
"tableName": ""
},
"offset": {
"segment": "0",
"snapshotCount": "",
"snapshotStartTime": ,
"tableName": "",
"totalSegments": "<0-5>"
}
},
{
// TypeC: This offset is logged once for each table and can be modified by the user.
"partition": {
"stream": ""
},
"offset": {
"cdc.start.position.timestamp": "", // required
"cdc.start.position.timestamp.format": "" // optional
}
}
]
```
### Update the offset
You can only edit the following offsets for a DynamoDB Stream.
```json
{
"partition": {
"stream": "" // required
},
"offset": {
"cdc.start.position.timestamp": "", // required
"cdc.start.position.timestamp.format": "" // optional
}
}
```
Considerations:
- You can change offset for each DynamoDB Stream individually. A task will
start streaming data from the specified timestamp from the configured
DynamoDB Stream.
- If no records in DynamoDB Streams are older than specified timestamp, all
records from DynamoDB Streams will be consumed again (equivalent to
`TRIM_HORIZON`).
- If timestamp specified is ahead of the most recent record in the corresponding
DynamoDB Stream, no existing records from stream will be consumed
(equivalent to `LATEST`).
- A new DynamoDB Table (leaseTable) will be created in the customer’s AWS
account on every offset modification after the connector goes into CDC phase.
- This is an asynchronous request. To check the status of this request, you
must use the check offset status API. For more information, see Get the
status of an offset request.
- Users are advised not to modify/edit any offsets of TypeA or TypeB. Users
must edit only TypeC offsets.
- If `cdc.start.position.timestamp.format` is not provided, it will default to
`yyyy-MM-dd'T'HH:mm:ss'Z'`. The value of the format must be a timestamp.
- You must configure `cdc.start.position.timestamp` in the format of
`cdc.start.position.timestamp.format` (configured or default). The value of the format must be a timestamp.
### JSON payload
The following table provides a description of the unique fields in the JSON payload
for managing offsets of the DynamboDB CDC Source connector.
| Field | Definition | Required/Optional |
|---------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------|
| `stream` | DynamoDB stream configured for CDC. | Required |
| `cdc.start.position.timestamp` | Used to specify the timestamp in the stream where a new application should start from. The value of the format must be a timestamp. | Required |
| `cdc.start.position.timestamp.format` | The format of timestamp in the stream where a new application should
start from. The format should abide by patterns specified in
[java.time.format.DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html#patterns).
The default is `yyyy-MM-dd'T'HH:mm:ss'Z'`. The value of the format must be a timestamp. | Optional |
## Connector events
This section describes the events the Amazon DynamoDB CDC connector creates
in different scenarios.
### Schemas and records created with the Avro converter
The Amazon DynamoDB CDC connector creates events the Avro converter can
leverage to manage schemas and records effectively.
Consider the following mapping and data representation:
- Topic name: `` (you can change this using the TopicRouting SMTs)
- Topic schema:
```json
[
"null",
{
"connect.name": "io.confluent.connect.dynamodb.iTable.Envelope",
"fields": [
{
"name": "op",
"type": "string"
},
{
"name": "ts_ms",
"type": "long"
},
{
"name": "ts_us",
"type": "long"
},
{
"name": "ts_ns",
"type": "long"
},
{
"default": null,
"name": "source",
"type": [
"null",
{
"connect.name": "io.confluent.connector.dynamodb.Source",
"fields": [
{
"default": null,
"name": "version",
"type": [
"null",
"string"
]
},
{
"name": "tableName",
"type": "string"
},
{
"name": "sync_mode",
"type": "string"
},
{
"name": "ts_ms",
"type": "long"
},
{
"name": "ts_us",
"type": "long"
},
{
"name": "ts_ns",
"type": "long"
},
{
"default": null,
"name": "snapshotStartTime",
"type": [
"null",
"long"
]
},
{
"default": null,
"name": "snapshotCount",
"type": [
"null",
"int"
]
},
{
"default": null,
"name": "segment",
"type": [
"null",
"int"
]
},
{
"default": null,
"name": "totalSegments",
"type": [
"null",
"int"
]
},
{
"default": null,
"name": "shard_id",
"type": [
"null",
"string"
]
},
{
"default": null,
"name": "seq_no",
"type": [
"null",
"string"
]
}
],
"name": "Source",
"namespace": "io.confluent.connector.dynamodb",
"type": "record"
}
]
},
{
"default": null,
"name": "before",
"type": [
"null",
{
"connect.name": "io.confluent.connect.dynamodb.iTable.Value",
"fields": [
{
"default": null,
"name": "document",
"type": [
"null",
"string"
]
}
],
"name": "Value",
"type": "record"
}
]
},
{
"default": null,
"name": "after",
"type": [
"null",
"Value"
]
}
],
"name": "Envelope",
"namespace": "io.confluent.connect.dynamodb.iTable",
"type": "record"
}
]
```
The following table gives a description of each of the fields names denoted in
the previous example.
| Field Name | Description |
|---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `schema` | The value’s schema, which describes the structure of the value’s payload.
For a particular table, all events will have the same schema |
| `connect.name` | The value’s schema, which describes the structure of the value’s payload.
For a particular table, all events will have the same schema |
| `op` | Mandatory string that describes the type of operation that caused the
connector to generate the event. Valid values are:
- `c`: create
- `u`: update
- `d`: delete
- `r`: read (applies to only snapshots) |
| `ts_ms`, `ts_us`, `ts_ns` | Field that displays the approximate timestamp at which the change event
was added to the DynamoDB Stream. In the `source` object, `ts_ms`
indicates the time at which the connector processed the event, based on
the system clock in the JVM running the Kafka Connect task. By
comparing the value for `payload.source.ts_ms` with the value for
`payload.ts_ms`, you can determine the lag between the DDB Stream and
the connector. `ts_ms` represents time in milliseconds, `ts_us`
represents time in microseconds, and `ts_ns` is the nanoseconds field.
However, `ts_ns` is not currently populated with true nanosecond-
precision values and may be derived from `ts_us` or `ts_ms`, so its
use is not recommended. |
| `source` | Mandatory field that describes the source metadata for the
event. This field contains information that you can use to compare this
event with other events, with regard to the origin of the events and the
order in which the events occurred. The source metadata includes:
- Source version (to denote evolution in source schema)
- Table name
- `SYNC_MODE`, valid values are:
- `CDC` (when in Streaming phase)
- `SNAPSHOT` (when in `SNAPSHOT` phase)
- Timestamp:
- `SNAPSHOT`: Timestamp at which table record was scanned.
- `CDC`: Approximate timestamp at which event was added into DDB Streams.
- `SNAPSHOT` fields:
- `snapshotStartTime`: Timestamp when snapshot was started for
table-segment which this event belongs to. Only present in SNAPSHOT
events.
- `snapshotCount`: Number of snapshot events scanned before this
event in this table-segment. Only present in SNAPSHOT events.
- `segment`: The table-segment to which this event belongs to. Only
present in SNAPSHOT events.
- `totalSegments`: Total number of table-segments for this table.
Only present in SNAPSHOT events
- `CDC` fields:
- `shard_id`: ShardId of the shard in DDB Stream from where this event
was consumed from. Only present in CDC events.
- `seq_no`: SequenceNumber of this event in shard in DDB Stream from
where this event was consumed from. Only present in CDC events. |
| `before` | An optional field that specifies the state of the row before the event
occurred. This field will be populated only with CDC events, when:
- `"op": "u"`
- `"op": "d"`
This field would only be populated when `StreamViewType` is
`NEW_AND_OLD_IMAGES` on DynamoDB Streams. |
| `after` | An optional field that specifies the state of the row after the event
occurred. This field will be populated with:
- `"op": "r"` (SNAPSHOT)
- `"op": "c"` (CDC)
- `"op": "u"` (CDC) |
| `connect.name` | `io.confluent.connect.dynamodb..Value` is the schema for the
payload’s `before` and `after` fields. This schema is specific to the table
(`tableName`)
Names of schemas for `before` and `after` fields are of the form
`io.confluent.connect.dynamodb..Value`, which ensures that the
schema name is unique in the database. This means that when using the
Avro converter, the resulting Avro schema for each table in each logical
source has its own evolution and history. |
The connector creates different kinds of events in the following scenarios:
### SNAPSHOT
Example record: `"op: r"`
```json
{
"op": "r",
"ts_ms": 1719830776204,
"ts_us": 1719830776204000,
"ts_ns": 1719830776204000000,
"source": {
"io.confluent.connector.dynamodb.Source": {
"version": null,
"tableName": "NumTable",
"sync_mode": "SNAPSHOT",
"ts_ms": 1719830776204,
"ts_us": 1719830776204000,
"ts_ns": 1719830776204000000,
"snapshotStartTime": {
"long": 1719830776087
},
"snapshotCount": {
"int": 1
},
"segment": {
"int": 0
},
"totalSegments": {
"int": 5
},
"shard_Id": null,
"seq_No": null
}
},
"before": null,
"after": {
"io.confluent.connect.dynamodb.NumTable.Value": {
"document": {
"string": "{\"Number2\":{\"N\":\"1\"},\"Number\":{\"N\":\"1\"}}"
}
}
}
}
```
### CDC
Example record 1: `"op: c"`
```json
{
"op": "c",
"ts_ms": 1719830940000,
"ts_us": 1719830940000000,
"ts_ns": 1719830940000000,
"source": {
"io.confluent.connector.dynamodb.Source": {
"version": null,
"tableName": "NumTable",
"sync_mode": "CDC",
"ts_ms": 1719830973951,
"ts_us": 1719830973951000,
"ts_ns": 1719830973951000000,
"snapshotStartTime": null,
"snapshotCount": null,
"segment": null,
"totalSegments": null,
"shard_Id": {
"string": "shardId-00000001719827428220-04b3c3e0"
},
"seq_No": {
"string": "000000000000000000247"
}
}
},
"before": null,
"after": {
"io.confluent.connect.dynamodb.NumTable.Value": {
"document": {
"string": "{\"Number2\":{\"N\":\"5\"},\"Number\":{\"N\":\"1\"}}"
}
}
}
}
```
Example record 2: `"op: u"`
```json
{
"op": "u",
"ts_ms": 1719830820000,
"ts_us": 1719830820000000,
"ts_ns": 1719830820000000,
"source": {
"io.confluent.connector.dynamodb.Source": {
"version": null,
"tableName": "NumTable",
"sync_mode": "CDC",
"ts_ms": 1719830875579,
"ts_us": 1719830875579000,
"ts_ns": 1719830875579000000,
"snapshotStartTime": null,
"snapshotCount": null,
"segment": null,
"totalSegments": null,
"shard_Id": {
"string": "shardId-00000001719827428220-04b3c3e0"
},
"seq_No": {
"string": "000000000000000000234"
}
}
},
"before": {
"io.confluent.connect.dynamodb.NumTable.Value": {
"document": {
"string": "{\"Number2\":{\"N\":\"3\"},\"Number\":{\"N\":\"1\"}}"
}
}
},
"after": {
"io.confluent.connect.dynamodb.NumTable.Value": {
"document": {
"string": "{\"Number2\":{\"N\":\"3\"},\"Number\":{\"N\":\"1\"},\"ONE\":{\"SS\":[\"NE\"]}}"
}
}
}
}
```
Example record 3: `"op: d"`
```json
{
"op": "d",
"ts_ms": 1719830820000,
"ts_us": 1719830820000000,
"ts_ns": 1719830820000000,
"source": {
"io.confluent.connector.dynamodb.Source": {
"version": null,
"tableName": "NumTable",
"sync_mode": "CDC",
"ts_ms": 1719830865036,
"ts_us": 1719830865036000,
"ts_ns": 1719830865036000000,
"snapshotStartTime": null,
"snapshotCount": null,
"segment": null,
"totalSegments": null,
"shard_Id": {
"string": "shardId-00000001719827428220-04b3c3e0"
},
"seq_No": {
"string": "000000000000000000232"
}
}
},
"before": {
"io.confluent.connect.dynamodb.NumTable.Value": {
"document": {
"string": "{\"Number2\":{\"N\":\"4\"},\"Number\":{\"N\":\"1\"},\"ONE\":{\"SS\":[\"NE\"]},\"ONE \":{\"SS\":[\"NE\",\"ONE\"]}}"
}
}
},
"after": null
}
```
## Quick start
Use this quick start to get up and running with the Amazon DynamoDB CDC
Source connector for Confluent Cloud.
### Prerequisites
Ensure you have the met the following requirements before moving ahead.
- Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS).
- The Confluent CLI installed and configured for the cluster. See [Install the
Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html).
- Authorized access to AWS and the Amazon DynamoDB database. For more
information, see [DynamoDB IAM policy](../cc-amazon-dynamo-db-sink.md#cc-dynamodb-policy).
- For networking considerations, see [Networking and DNS](../overview.md#connect-internet-access-resources).
To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](../static-egress-ip.md#cc-static-egress-ips).
- Kafka cluster credentials. The following lists the different ways you can provide credentials.
- Enter an existing [service account](../service-account.md#s3-cloud-service-account) resource ID.
- Create a Confluent Cloud [service account](../service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](../service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements.
- Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.
### Using the Confluent Cloud Console
#### Step 1: Launch your Confluent Cloud cluster
To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../../get-started/index.md#cloud-create-kafka-cluster).
#### Step 2: Add a connector
In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add
connector**.
#### Step 3: Select your connector
Click the **DynamoDB CDC Source** connector card.
#### Step 4: Enter the connector details
#### NOTE
* Ensure you have all your [prerequisites](#cc-amazon-dynamodb-cdc-source-prereqs) completed.
* An asterisk ( \* ) designates a required entry.
At the **Add Amazon DynamoDB CDC Source Connector** screen, complete the
following:
### Kafka access
1. Select the way you want to provide **Kafka Cluster credentials**. You can
choose one of the following options:
- **My account**: This setting allows your connector to globally access everything
that you have access to. With a user account, the connector uses an API key and
secret to access the Kafka cluster. This option is not recommended for production.
- **Service account**: This setting limits the access for your connector by using a
[service account](../service-account.md#s3-cloud-service-account). This option is recommended for
production.
- **Use an existing API key**: This setting allows you to specify an API key and a
secret pair. You can use an existing pair or create a new one. This method is not
recommended for production environments.
#### NOTE
Freight clusters support only service accounts for Kafka authentication.
2. Click **Continue**.
### Authentication
1. Configure the authentication properties:
**Authentication method**
- **Authentication method**: Select how you want to authenticate with AWS.
* If you select **Access Keys**, enter your AWS credentials in the **Amazon Access Key ID** and **Amazon Secret Access Key fields**. For information about how to set these up, [Access Keys](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html).
* If you select **IAM Roles**, choose an existing integration name under Provider integration name dropdown that has access to your resource. For more information, see [Manage Provider Integration for Fully Managed Connectors in Confluent Cloud](../provider-integration.md#cloud-pi-quickstart).
- **Use secret manager**: Fetch sensitive configuration values from a secret manager.
**AWS credentials**
- **Provider Integration**: Select an existing integration that has access to your resource if you choose **IAM Roles** as your authentication method.
- **Amazon Access Key ID**: Specify the Amazon Access Key ID to let this connector access DynamoDB if you choose **Access Kyes** as the authentication method.
- **Amazon Secret Access Key**: Specify the Amazon Secret Access Key to let this connector access DynamoDB if you choose **Access Keys** as the authentication method.
**Secret manager configuration**
- **Secret manager**: Select the secret manager to use for retrieving sensitive data.
- **Configurations from Secret manager**: Select the configurations whose values Confluent Cloud should
fetch from the secret manager.
- **Provider Integration**: Select an existing integration that has access to your resource if you choose **IAM Roles** as your authentication method.
**DynamoDB Details**
- **AWS DynamoDB API Endpoint**: Provide the AWS DynamoDB API endpoint.
2. Click **Continue**.
### Configuration
Note that configuration properties that are not shown in the
Cloud Console use the default values. For all property values and
definitions, see [Configuration properties](#cc-amazon-dynamodb-cdc-source-config-properties).
**Output messages**
- **Select output record value format**: Sets the output Kafka record value format. Valid entries are `AVRO`, `JSON_SR`, or `PROTOBUF`. A valid schema must be available in [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) to use a schema-based message format (for example, Avro, JSON Schema, or Protobuf).
- **Output Kafka record key format**: Sets the output Kafka record key format. Valid entries are `AVRO`, `JSON_SR`, or `PROTOBUF`. A valid schema must be available in [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) to use a schema-based message format (for example, Avro, JSON Schema, or Protobuf).
**Schema Validation**
- **Document content format**: Sets the format for the document content. Valid values are `json_string` and `schema`. The json_string mode (default) outputs records as JSON strings in a document field. The schema mode infers Kafka Connect schemas from DynamoDB records and outputs structured data with individual fields. Note that schema mode requires appropriate converters (for example, Avro or JSON_SR) for full functionality.
- **Document schema source**: Specifies the source of the document schema. This configuration is only used when **Document content format** is set to `schema`.
* `user_provided` (default): Records are validated against the schema specified in the **Document Schema Value** configuration.
* `auto_infer`: The connector infers the schema from incoming records and registers it in the Schema Registry.
#### WARNING
Using `auto_infer` for tables with evolving or variable schemas is not recommended. Schema evolution can result in multiple schema versions being registered, which might exceed registry limits or cause operational issues. In addition, the connector might fail to register new schemas if they are not backward compatible. As a workaround, you can set the Schema Compatibility mode to `NONE` in Confluent Cloud, but you must carefully assess the impact of incompatible schema versions on downstream consumers before making this change.
- **Document Schema Value**: Specifies an Avro schema definition for the document field of the SourceRecord. This is required when **Document schema source** is set to `user_provided`. Example: `{"type":"record","name":"User","fields":[{"name":"id","type":["null","string"]},{"name":"name","type":["null","string"]}]}`.
- **Document Schema Validation Mode**: Specifies the schema validation mode.
* `strict`: Fails if extra fields, missing mandatory fields, or type mismatches are detected.
* `lenient`: Drops extra fields and continues processing, but fails if missing mandatory fields or type mismatches (data integrity issues) are detected.
- **Decimal Handling Mode**: Controls how DynamoDB Number types are mapped to Kafka Connect schemas during auto-inference. This setting is applicable only when **Document content format** is set to `schema` and **Document schema source** is set to `auto_infer`. Options available are:
* `double` (default): Always maps to `FLOAT64`. This is the recommended setting for schema consistency and backward compatibility.
* `long`: Always maps to `INT64`.
* `string`: Always maps to `STRING`
* `auto`: Infers `INT32`/`INT64`/`FLOAT64` based on value.
#### WARNING
Using `auto` mode can cause backward-incompatible schemas if the range of numeric values varies widely between messages. The same field might be inferred as different types (for example, `INT32` versus `INT64` versus `FLOAT64`), leading to Schema Registry registration failures.
**DynamoDB Details**
- **DynamoDB Table Sync Mode**: Select a table sync mode from the **DynamoDB Table Sync Mode** dropdown
list. Valid values are:
- `CDC`: Perform CDC only.
- `SNAPSHOT`: Perform a snapshot only.
- `SNAPSHOT_CDC` (Default): The connector starts with a
snapshot and then switches to CDC mode upon completion.
- **Table Discovery Mode**: Select a table discovery mode from the **Table Discovery Mode**
dropdown list. Valid values are:
- `INCLUDELIST`: A comma-separated list of DynamoDB table names to be
captured. This is required if `dynamodb.table.discovery.mode` is
set to `INCLUDELIST`.
- `TAG`: A semi-colon-separated list of pairs in the form
`:,` that is used to create tag filters.
For example, `key1:v1,v2;key2:v3,v4` will include all tags that
match `key1` key with value of either `v1` or `v2`, and match
`key2` with value of either `v3` or `v4`. Any `keys` not
specified will be excluded.
- **AWS Resource Group Tag API Endpoint**: AWS Resource Group Tag API Endpoint. Required if `dynamodb.table.discovery.mode` is selected as TAG.
- **DynamoDB Table Tag Filters**: A semi-colon-separated list of pairs in the form `:,` that is used to create tag filters. For example, `key1:v1,v2;key2:v3,v4` will include all tags that match `key1` key with value of either `v1` or `v2`, and match `key2` with value of either `v3` or `v4`. Any `keys` not specified will be excluded.
- **Tables Include List**: In the **Tables Include List** field, enter a comma-separated list of
DynamoDB table names to be captured. Note that this is required if
`dynamodb.table.discovery.mode` is set to `INCLUDELIST`.
**CDC Details**
- **CDC Initial Position in Stream**: Specifies the position in the stream where a new application should start from. This is used during initial application
bootstrap (when a checkpoint doesn’t exist for a shard or its parents). Used only in CDC mode. Possible values:
* `TRIM_HORIZON` - Start from the oldest available data record.
* `LATEST` - Start after the most recent data record (fetch new data).
* `AT_TIMESTAMP` - Start from the record at or after the specified server-side timestamp.
- **CDC Initial Position in Stream Timestamp**: Specifies the timestamp in the stream where a new application should start from. This is used during initial application bootstrap (when a checkpoint doesn’t exist for a shard or its parents). Used only in CDC mode with `AT_TIMESTAMP` selected in the **CDC Initial Position in Stream** field.
- **CDC Initial Position in Stream Timestamp Format**: The format of timestamp in the stream where a new application should start from. The format should abide by patterns specified in [java.time.format.DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html#patterns). This is used during initial application bootstrap (when a checkpoint doesn’t exist for a shard or its parents). Used only in CDC mode with `AT_TIMESTAMP` selected in the **CDC Initial Position in Stream** field.
### **Show advanced configurations**
- **Schema context**: Select a schema context to use for this connector, if using
a schema-based data format. This property defaults to the **Default** context,
which configures the connector to use the default schema set up for Schema Registry in your
Confluent Cloud environment. A schema context allows you to use separate schemas (like
schema sub-registries) tied to topics in different Kafka clusters that share the
same Schema Registry environment. For example, if you select a non-default context, a
**Source** connector uses only that schema context to register a schema and a
**Sink** connector uses only that schema context to read from. For more
information about setting up a schema context, see [What are schema contexts and when should you use them?](../../sr/faqs-cc.md#faq-schema-contexts).
- **Key Converter Reference Subject Name Strategy**: Set the subject reference name strategy for key. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. Note that the subject reference name strategy can be selected only for `PROTOBUF` format with the default strategy being `DefaultReferenceSubjectNameStrategy`.
**Additional Configs**
- **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`.
- **Errors Tolerance**: Use this property to configure the connector’s error handling behavior.
#### WARNING
Use this property with caution for sink connectors, as it can lead to data loss. If you set this property to `all`, the connector does not fail on errant records, but logs them (and sends to DLQ for sink connectors) and continues processing. If you set this property to `none`, the connector task fails on errant records.
- **Value Converter Ignore Default For Nullables**: When set to `true`, this property ensures that the corresponding record in Kafka is `null`, instead of showing the default column value. Applies to the `AVRO`, `PROTOBUF`, and `JSON_SR` converters.
- **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals:
`BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and
`NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value.
- **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
- **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters.
- **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
- **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration.
- **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
**Auto-restart policy**
- **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its
task in the event of user-actionable errors. Defaults to `true`, enabling the connector to
automatically restart in case of user-actionable errors. Set this property to `false` to
disable auto-restart for failed connectors. If disabled, you must manually restart the connector.
**CDC Details**
- **Prefix for CDC Checkpointing table**: Prefix for CDC Checkpointing
tables, must be unique per connector. Checkpointing table is used to
store the last processed record for each shard and is used to resume
from last processed record in case of connector restart. This is
applicable only in CDC mode.
- **CDC Checkpointing Table Billing Mode**: Define billing mode for
internal checkpoint table created with CDC. Valid values are
`PROVISIONED` and `PAY_PER_REQUEST`. Default is `PROVISIONED`.
Use `PAY_PER_REQUEST` for unpredictable application traffic and
on-demand billing mode. Use `PROVISIONED` for predictable
application traffic and provisioned billing mode.
- **Max records per DDB Streams Poll**: Maximum number of records that can be returned in single DynamoDB Stream getRecords
- **Read capacity for CDC Checkpointing table**: Read capacity for CDC checkpointing table. The checkpointing table is used to track the leases (shards) of the DynamoDB tables and helps determine the resume position in case of connector restarts.
- **Write capacity for CDC Checkpointing table**: Write capacity for CDC checkpointing table. The checkpointing table is used to track the leases (shards) of the DynamoDB tables and helps determine the resume position in case of connector restarts.
**Snapshot Details**
- **Max records per Table Scan**: Maximum number of records that can be
returned in single DynamoDB read operation. Only applicable to
`SNAPSHOT` phase. Note that there is 1 MB size limit as well.
- **Snapshot Table RCU consumption percentage**: Configure the percentage
of table read capacity that will be used as a maximum limit of RCU
consumption rate.
**DynamoDB Details**
- **Maximum batch size**: The maximum number of records the connector
will wait for before publishing the data on the topic. The connector
may still return fewer records if no additional records are
available.
- **Poll linger milliseconds**: The maximum time to wait for a record
before returning an empty batch. The default is 5 seconds.
**Transforms**
- **Single Message Transformations**: To add a new SMT, see [Add transforms](../single-message-transforms.md#cc-single-message-transforms-ui).
For more information about unsupported SMTs, see
[Unsupported transformations](../single-message-transforms.md#cc-single-message-transforms-unsupported-transforms).
**Processing position**
- **Set offsets**: Click **Set offsets** to define a specific offset for
this connector to begin procession data from. For more information
on managing offsets, see [Manage offsets](../offsets.md#connect-custom-offsets).
For all property values and definitions, see
[Configuration properties](#cc-amazon-dynamodb-cdc-source-config-properties).
- Click **Continue**.
### Sizing
Based on the number of topic partitions you select, you will be provided
with a recommended number of tasks.
1. To change the number of recommended tasks, enter the number of
[tasks](/platform/current/connect/concepts.html#tasks) for the connector to use in
the **Tasks** field.
2. Click **Continue**.
### Review and Launch
1. Verify the connection details.
2. Click **Launch**.
The status for the connector should go from **Provisioning** to
**Running**.
#### Step 5: Check the Kafka topic.
After the connector is running, verify that messages are populating your Kafka
topic.
For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section.
### Using the Confluent CLI
You can use the following steps to set up and run the connector using the
Confluent CLI, but first, ensure you have met all [prerequisites](#cc-amazon-dynamodb-cdc-source-config-properties).
#### Step 1: List the available connectors
Enter the following command to list available connectors:
```none
confluent connect plugin list
```
#### Step 2: List the connector configuration properties
Enter the following command to show the connector configuration properties:
```none
confluent connect plugin describe
```
The command output shows the required and optional configuration properties.
#### Step 3: Create the connector configuration file
Create a JSON file that contains the connector configuration properties. The
following example shows required and optional connector properties.
```json
{
"config": {
"connector.class": "DynamoDbCdcSource",
"name": "DynamoDbCdcSourceConnector_0",
"kafka.auth.mode": "KAFKA_API_KEY",
"kafka.api.key": "",
"kafka.api.secret": "",
"schema.context.name": "default",
"aws.access.key.id": "",
"aws.secret.access.key": "",
"output.data.format": "JSON",
"dynamodb.service.endpoint": "http://localhost:8080",
"dynamodb.table.discovery.mode": "INCLUDELIST",
"dynamodb.table.sync.mode": "SNAPSHOT_CDC",
"dynamodb.table.includelist": "t1, t2",
"max.batch.size": "1000",
"poll.linger.ms": "5000",
"dynamodb.snapshot.max.poll.records": "1000",
"dynamodb.cdc.checkpointing.table.prefix": "connect-KCL-",
"dynamodb.cdc.table.billing.mode": "PROVISIONED",
"dynamodb.cdc.max.poll.records": "5000",
"tasks.max": "1"
}
}
```
For all property values and definitions, see
[Configuration properties](#cc-amazon-dynamodb-cdc-source-config-properties).
**SMTs**: For details about adding SMTs using the
Confluent CLI, see the [Single Message Transformations](../single-message-transforms.md#cc-single-message-transforms) documentation. For a list of SMTs that are not
supported with this connector, see
[Unsupported transformations](../single-message-transforms.md#cc-single-message-transforms-unsupported-transforms).
#### Step 4: Load the configuration file and create the connector
Enter the following command to load the configuration and start the connector:
```none
confluent connect cluster create --config-file .json
```
For example:
```none
confluent connect cluster create --config-file dynamodb-cdc-source-config.json
```
Example output:
```none
Created connector DynamoDbCdcSourceConnector_0 lcc-ix4dl
```
#### Step 5: Check the connector status
Enter the following command to check the connector status:
```none
confluent connect cluster list
```
Example output:
```none
ID | Name | Status | Type
+-----------+-----------------------------+---------+------+
lcc-ix4dl | DynamoDbCdcSourceConnector_0| RUNNING | source
```
#### Step 6: Check the Kafka topic.
After the connector is running, verify that messages are populating your Kafka
topic.
For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section.
## Configuration properties
Use the following configuration properties with the fully managed connector.
### How should we connect to your data?
`name`
: Sets a name for your connector.
* Type: string
* Valid Values: A string at most 64 characters long
* Importance: high
### Kafka Cluster credentials
`kafka.auth.mode`
: Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string
* Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY
* Importance: high
`kafka.api.key`
: Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password
* Importance: high
`kafka.service.account.id`
: The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string
* Importance: high
`kafka.api.secret`
: Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password
* Importance: high
### Schema Config
`schema.context.name`
: Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
* Type: string
* Default: default
* Importance: medium
`key.converter.reference.subject.name.strategy`
: Set the subject reference name strategy for key. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string
* Default: DefaultReferenceSubjectNameStrategy
* Importance: medium
### Authentication method
`authentication.method`
: Select how you want to authenticate with AWS.
* Type: string
* Default: Access Keys
* Importance: high
`secret.manager.enabled`
: Fetch sensitive configuration values from a secret manager.
* Type: boolean
* Default: false
* Importance: high
### Secret manager configuration
`secret.manager`
: Select the secret manager to use for retrieving sensitive data.
* Type: string
* Importance: high
`secret.manager.managed.configs`
: Select the configurations to fetch their values from the secret manager.
* Type: list
* Importance: high
`secret.manager.provider.integration.id`
: Select an existing provider integration that has access to your secret manager.
* Type: string
* Importance: high
### AWS credentials
`aws.access.key.id`
: * Type: password
* Importance: high
`aws.secret.access.key`
: * Type: password
* Importance: high
`provider.integration.id`
: Select an existing integration that has access to your resource. In case you need to integrate a new IAM role, use provider integration
* Type: string
* Importance: high
### Output messages
`output.data.format`
: Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Please configure Confluent Cloud Schema Registry.
* Type: string
* Default: AVRO
* Importance: high
`output.data.key.format`
: Sets the output Kafka record key format. Valid entries are AVRO, JSON_SR, PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
* Type: string
* Default: AVRO
* Valid Values: AVRO, JSON_SR, PROTOBUF
* Importance: high
### Schema Validation
`document.content.format`
: Sets the document content format. ‘json_string’ mode (default) outputs records as JSON strings in a ‘document’ field. ‘schema’ mode infers Kafka Connect schemas from DynamoDB records and outputs structured data with individual fields. Note that ‘schema’ mode requires appropriate converters (AVRO, JSON_SR, etc.) for full functionality.
* Type: string
* Default: json_string
* Valid Values: json_string, schema
* Importance: high
`document.schema.source`
: Specifies the source of the document schema. This configuration is only used when ‘document.content.format’ is set to ‘schema’. ‘user_provided’ (default) - records are validated against the schema specified in ‘schema.user.schema’ configuration. ‘auto_infer’ - the connector infers the schema from incoming records and registers it in Schema Registry.
WARNING: Using ‘auto_infer’ for tables with evolving or variable schemas is not recommended. Schema evolution may result in multiple schema versions being registered in Schema Registry, which can exceed registry limits or lead to operational issues. In addition, the connector may fail to register new schemas if they are not backward compatible with previously registered versions. As a workaround, you may set the Schema Compatibility mode to NONE in Confluent Cloud; however, carefully assess the impact of incompatible schema versions on downstream consumers before making this change.
* Type: string
* Default: user_provided
* Valid Values: auto_infer, user_provided
* Importance: medium
`document.schema.value`
: Specifies an Avro schema definition for the document field of the SourceRecord. Required when ‘document.schema.source’ is set to ‘user_provided’. Example: {“type”:”record”,”name”:”User”,”fields”:[{“name”:”id”,”type”:[“null”,”string”]},{“name”:”name”,”type”:[“null”,”string”]}]}.
* Type: string
* Importance: medium
`document.schema.validation`
: Schema validation mode. ‘strict’ = fail if extra fields, missing mandatory fields, or type mismatches are detected. ‘lenient’ = drop extra fields and continue processing, but still fail on missing mandatory fields and type mismatches (data integrity issues).
* Type: string
* Default: strict
* Valid Values: lenient, strict
* Importance: medium
`decimal.handling.mode`
: Controls how DynamoDB Number types are mapped to Kafka Connect schemas during auto-inference. Options: ‘double’ (always FLOAT64 - RECOMMENDED), ‘long’ (always INT64), ‘string’ (always STRING), ‘auto’ (infers INT32/INT64/FLOAT64 based on value). Only applicable when ‘document.content.format=schema’ and ‘document.schema.source=auto_infer’. Default is ‘double’ for schema consistency and backward compatibility. WARNING: Using ‘auto’ mode can cause backward incompatible schemas when the range of numeric values varies widely between messages, as the same field may be inferred as different types (e.g., INT32 vs INT64 vs FLOAT64), leading to Schema Registry registration failures.
* Type: string
* Default: double
* Valid Values: auto, double, long, string
* Importance: medium
### DynamoDB Details
`dynamodb.service.endpoint`
: AWS DynamoDB API Endpoint
* Type: string
* Importance: high
`dynamodb.table.sync.mode`
: Define table sync mode. SNAPSHOT_CDC - start with SNAPSHOT and switch to CDC mode on completion; SNAPSHOT - perform SNAPSHOT only,CDC - perform CDC only
* Type: string
* Default: SNAPSHOT_CDC
* Importance: high
`dynamodb.table.discovery.mode`
: Specifies the table discovery mode that defines the list of tables to be captured.
TAG - use resource tag discovery using AWS Resource Tagging endpoint.
> INCLUDELIST- use explicit table names to be included
* Type: string
* Default: INCLUDELIST
* Importance: high
`aws.resource.tagging.endpoint`
: AWS Resource Group Tag API Endpoint. Required if dynamodb.table.discovery.mode is selected as TAG
* Type: string
* Importance: high
`dynamodb.table.tag.filters`
: A semi-colon-separated list of pairs in the form :, that is used to create tag filters. For example, key1:v1,v2;key2:v3,v4 will include all tags that match key1 key with value of either v1 or v2, and match key2 with value of either v3 or v4. Any keys not specified will be excluded.
* Type: string
* Importance: high
`dynamodb.table.includelist`
: A comma-separated list of DynamoDB table names to be captured. This is required if dynamodb.table.discovery.mode is set to INCLUDELIST
* Type: string
* Importance: high
`max.batch.size`
: The maximum number of records that will be returned by the connector to Connect. The connector may still return fewer records if no additional records are available.
* Type: int
* Default: 1000
* Valid Values: [100,…,10000]
* Importance: medium
`poll.linger.ms`
: The maximum time to wait for a record before returning an empty batch. The call to poll can return early before `poll.linger.ms` expires if `max.batch.size` records are received.
* Type: long
* Default: 5000 (5 seconds)
* Valid Values: [0,…,300000]
* Importance: medium
### Snapshot Details
`dynamodb.snapshot.max.poll.records`
: Maximum number of records that can be returned in single DynamoDB read operation. Only applicable to SNAPSHOT phase. Note that there is 1MB size limit as well.
* Type: int
* Default: 1000
* Importance: medium
`dynamodb.snapshot.max.rcu.percentage`
: Configure percentage of table read capacity that will be used as a maximum limit of RCU consumption rate
* Type: int
* Default: 50
* Importance: medium
### CDC Details
`dynamodb.cdc.initial.stream.position`
: Specifies the position in the stream where a new application should start from. This is used during initial application bootstrap (when a checkpoint doesn’t exist for a shard or its parents). Used only in CDC mode.TRIM_HORIZON - Start from the oldest available data record. LATEST - Start after the most recent data record (fetch new data).AT_TIMESTAMP - Start from the record at or after the specified server-side timestamp
* Type: string
* Importance: medium
`cdc.start.position.timestamp`
: Specifies the timestamp in the stream where a new application should start from. This is used during initial application bootstrap (when a checkpoint doesn’t exist for a shard or its parents). Used only in CDC mode.
* Type: string
* Importance: medium
`cdc.start.position.timestamp.format`
: The format of timestamp in the stream where a new application should start fromThe format should abide by patterns specified in java.time.format.DateTimeFormatter [https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html#patterns](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html#patterns). This is used during initial application bootstrap (when a checkpoint doesn’t exist for a shard or its parents). Used only in CDC mode.
* Type: string
* Default: yyyy-MM-dd’T’HH:mm:ss’Z’
* Importance: medium
`dynamodb.cdc.checkpointing.table.prefix`
: Prefix for CDC Checkpointing tables, must be unique per connector. Checkpointing table is used to store the last processed record for each shard and is used to resume from lastprocessed record in case of connector restart. This is applicable only in CDC mode.
* Type: string
* Default: connect-KCL-
* Importance: medium
`dynamodb.cdc.table.billing.mode`
: Define billing mode for internal checkpoint table created with CDC. Allowed values: PROVISIONED, PAY_PER_REQUEST.Default is PROVISIONED. Use PAY_PER_REQUEST for unpredictable application traffic and on-demand billing mode. Use PROVISIONED for predictable application traffic and provisioned billing mode.
* Type: string
* Default: PROVISIONED
* Importance: medium
`dynamodb.cdc.max.poll.records`
: Maximum number of records that can be returned in single DynamoDB Stream getRecords
* Type: int
* Default: 5000
* Importance: medium
`dynamodb.cdc.checkpointing.table.read.capacity`
: Read capacity for CDC checkpointing table. The checkpointing table is used to track the leases (shards) of the DynamoDB tables and helps determine the resume position in case of connector restarts.
* Type: int
* Default: 50
* Importance: low
`dynamodb.cdc.checkpointing.table.write.capacity`
: Write capacity for CDC checkpointing table. The checkpointing table is used to track the leases (shards) of the DynamoDB tables and helps determine the resume position in case of connector restarts.
* Type: int
* Default: 50
* Importance: low
### Number of tasks for this connector
`tasks.max`
: Maximum number of tasks for the connector.
* Type: int
* Valid Values: [1,…]
* Importance: high
### Additional Configs
`header.converter`
: The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string
* Importance: low
`producer.override.compression.type`
: The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
* Type: string
* Importance: low
`producer.override.linger.ms`
: The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: [https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms).
* Type: long
* Valid Values: [100,…,1000]
* Importance: low
`value.converter.allow.optional.map.keys`
: Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
* Type: boolean
* Importance: low
`value.converter.auto.register.schemas`
: Specify if the Serializer should attempt to register the Schema.
* Type: boolean
* Importance: low
`value.converter.connect.meta.data`
: Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean
* Importance: low
`value.converter.enhanced.avro.schema.support`
: Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
* Type: boolean
* Importance: low
`value.converter.enhanced.protobuf.schema.support`
: Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.flatten.unions`
: Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.generate.index.for.unions`
: Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.generate.struct.for.nulls`
: Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.int.for.enums`
: Whether to represent enums as integers. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.latest.compatibility.strict`
: Verify latest subject version is backward compatible when use.latest.version is true.
* Type: boolean
* Importance: low
`value.converter.object.additional.properties`
: Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
* Type: boolean
* Importance: low
`value.converter.optional.for.nullables`
: Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.optional.for.proto2`
: Whether proto2 optionals are supported. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.scrub.invalid.names`
: Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.use.latest.version`
: Use latest version of schema in subject for serialization when auto.register.schemas is false.
* Type: boolean
* Importance: low
`value.converter.use.optional.for.nonrequired`
: Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
* Type: boolean
* Importance: low
`value.converter.wrapper.for.nullables`
: Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`value.converter.wrapper.for.raw.primitives`
: Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
* Type: boolean
* Importance: low
`errors.tolerance`
: Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
* Type: string
* Default: none
* Importance: low
`key.converter.key.schema.id.serializer`
: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
* Type: string
* Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
* Importance: low
`key.converter.key.subject.name.strategy`
: How to construct the subject name for key schema registration.
* Type: string
* Default: TopicNameStrategy
* Importance: low
`value.converter.decimal.format`
: Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string
* Default: BASE64
* Importance: low
`value.converter.flatten.singleton.unions`
: Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
* Type: boolean
* Default: false
* Importance: low
`value.converter.ignore.default.for.nullables`
: When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.
* Type: boolean
* Default: false
* Importance: low
`value.converter.reference.subject.name.strategy`
: Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string
* Default: DefaultReferenceSubjectNameStrategy
* Importance: low
`value.converter.value.schema.id.serializer`
: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
* Type: string
* Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
* Importance: low
`value.converter.value.subject.name.strategy`
: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string
* Default: TopicNameStrategy
* Importance: low
### Auto-restart policy
`auto.restart.on.user.error`
: Enable connector to automatically restart on user-actionable errors.
* Type: boolean
* Default: true
* Importance: medium
## Frequently Asked Questions
The frequently asked questions (FAQs) address common questions and issues encountered while using the fully managed Amazon
DynamoDB CDC Source connector for Confluent Cloud.
#### NOTE
To illustrate the concepts discussed in the FAQs, consider a scenario with the following example:
* Resources: 10 Tables
* Workload: 1 Task that processes data from all 10 tables
* Connector starts at: `t0`
* Snapshot completion for Tables T1,…T10: t1,…t10
This example will be referenced throughout the FAQs to provide context for various questions and answers.
### Snapshot phase issues
**Problem**: The connector failed with the error: The earliest snapshot has taken more than 24 hours in SNAPSHOT_CDC mode,
causing streams to expire. Increase the number of tasks to complete the snapshot within 24 hours.
**Reason**: During the snapshot phase, each task first completes a full snapshot of all the DynamoDB tables assigned
to it before starting to capture change data. If the total time taken to snapshot all tables exceeds the DynamoDB Streams retention period of 24 hours, the connector will be unable to start CDC from the
beginning of the stream for tables such as T1 whose snapshot completed later.
**Solution**: To expedite the snapshot process and avoid stream expiration:
- **Task assignment**: Ensure that one table is assigned to a maximum of one task during the snapshot phase.
This prevents a single task from spending excessive time snapshotting multiple tables.
* **Increase task count (during snapshot)**: Temporarily increase the number of tasks in your connector configuration
to match the number of tables. For example, if you have 10 tables, use 10 tasks for the snapshot.
* **Reduce task count (post-snapshot)**: Once the snapshot phase is complete, you can reduce the number of tasks to
your desired level for the CDC phase.
* **Optimize read capacity units (RCUs)**: Adjust the RCU provisioned for your DynamoDB tables to improve snapshot performance.
* **Manage snapshot RCU percentage**: Use the `dynamodb.snapshot.max.rcu.percentage` configuration to control the
maximum percentage of the table’s RCU that the connector can use for snapshots.
* **Review producer configuration**: Check your Logical Kafka Cluster (LKC), cluster type, and networking settings to
ensure they are not bottlenecks.
* **Set max poll records**: Configure `dynamodb.snapshot.max.poll.records` to manage the maximum number of records
fetched per poll during the snapshot. Note that AWS limits each request to 1 MB.
- **Exclude Problematic Tables (If Necessary)**: If the above recommendations do not resolve the issue, consider
removing the tables from the connector’s include_list.
**Example**: For five tables, configure five tasks for the initial snapshot. After the snapshot completes, you can
reduce the task count during the CDC phase. Be aware that there might be a heavy lag during the initial CDC phase
as it catches up on changes that occurred during the snapshot.
#### What happens when a new table is added to an existing connector that is already in the CDC phase?
When you add a new table to a connector that is currently processing other tables in CDC phase (for example,
4 tables with 1 task), the following occurs:
- The existing task will now be responsible for the newly added table in addition to the existing ones.
- The connector will first perform a full snapshot of the newly added table.
- CDC will only begin for the newly added table once its snapshot is complete.
#### Is it recommended to create new connectors to handle different tables?
Confluent recommends creating separate connectors for larger tables. For smaller tables, you can often increase
the task count within the same connector.
#### How can customers know whether the snapshot is complete? Are any metrics available?
You can observe the offsets on the Confluent Cloud Console. Once a snapshot for a table
is complete, you will see an initial offset for that table. Currently, there are no
specific metrics to explicitly indicate snapshot completion.
#### There’s no throughput in the CDC phase, how to debug this?
To troubleshoot a lack of throughput during the CDC phase, first check the Write throughput on the
DynamoDB table in the AWS console. If there are no writes to the DynamoDB table, there will be no
events for the connector to process. If writes are occurring, you can use **Custom Offset** to rewind
the connector to a suitable timestamp. This will create a new checkpointing table and initiate a fresh CDC start.
Common limitations
: - **Task-to-table mapping**: A single DynamoDB table can only be processed by one task at any given time.
Configuring more tasks than the number of tables will result in idle tasks. The maximum number of tasks
you should configure is equal to the total number of tables being processed by the connector.
- **dynamodb.snapshot.max.poll.records**: This configuration limits the number of records returned in a
single DynamoDB read operation during the snapshot. Each read operation has a maximum size of 1 MB.
- **dynamodb.cdc.max.poll.records**: This setting tunes the maximum number of records fetched in a single
`getRecords` operation from the DynamoDB Stream. Each `getRecords` operation also has a 1 MB size limit.
- **DynamoDB stream consumers**: AWS DynamoDB Streams has a limitation of a maximum of two consumers per stream.
Since the Confluent connector acts as one consumer, only one additional consumer can be active on the same stream.
- **Provisioned throughput (auto-scaling Off)**: If a table’s pricing mode is provisioned with auto-scaling disabled,
the connector’s throughput will be limited by the RCU configured on the table.
- **Snapshot duration limit**: A task might fail if the snapshot of a table takes longer than 24 hours because
CDC items in the stream expire after this period. To prevent this, ensure an appropriate task-to-table ratio and sufficient table RCU.
#### Is there a way to restart the connector from the Snapshot checkpoint to avoid populating the Kafka topic with duplicate data?
To ensure no duplicate data, you need to start from the beginning. This involves deleting the existing Kafka
topics for the affected tables and then restarting the connector, which will initiate a new snapshot.
### Failure metrics
#### Are there any metrics to check for snapshot failure?
Currently, there are no specific metrics dedicated to snapshot failure. The error message in the connector logs
and email notifications for task failures are the primary indicators.
However, if you have a 1:1 task-to-table mapping during the snapshot phase, you can monitor the [topic throughput
metric](../../monitoring/metrics-api-examples.md#metrics-api-bytes-received). A drop in throughput for a specific topic could indicate that the snapshot
or CDC process for that table has stopped. If a task is handling multiple tables, a temporary drop in throughput for a
table’s topic might occur after its snapshot completes and the task begins snapshotting another table.
### Estimating snapshot completion time
#### How to evaluate the time it would take to complete a snapshot of each table?
There are no pre-estimation metrics available within the connector. The snapshot duration depends on several
factors that you need to evaluate:
- Record size: Larger records will take longer to transfer.
- Table size: Tables with more data will require more time to snapshot.
- RCU: The RCUs provisioned for the table directly impact the read throughput during the snapshot.
- Throughput metric: Once the connector is running, you can observe the throughput metric to get an idea of the
snapshot speed, but this is only available during the process.
### 24-Hour limit configuration
#### Is the 24-hour limit configurable?
No, the 24-hour limit for DynamoDB streams data retention is a limitation imposed by AWS DynamoDB.
For more information, see [Data retention limit for DynamoDB Streams](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html).
### Table stream and task load
#### A DDB table stream can have at most one task. What happens if one task cannot handle the load due to the high throughput of changes?
The current design assigns one task per table. If a single task cannot keep up
with a high volume of changes on a table’s stream, it will lead to limited throughput and an increasing lag.
During the streams (CDC) phase, the task automatically scales its consumption capacity based on the number of
active shards in the streams it is consuming from, up to a limit set by Confluent.
### Threads per task
#### Are these threads running concurrently?
Tasks operate serially but multiple consumers work in parallel for each table.
#### Is there any partitioning or ordering of stream events that consumers should be aware of, due to tasks?
No, all records will appear in the same order as they were added to the DDB Stream. The connector processes
the records in order, ensuring that the DDB Stream’s order of events is maintained.
### Monitoring task/connector performance
#### How to monitor the number of messages processed per task/connector?
You can monitor the following metrics:
- Throughput: The overall rate at which data is being processed.
- Records/sec: The number of DynamoDB records being processed per second.
#### How to monitor the lag in a task, especially when it is processing multiple tables?
Query the metric `io.confluent.kafka.connect/dynamodb_cdc_source_connector_lag_milliseconds`.
This metric represents the time difference (in milliseconds) between the timestamp of the
last change event in the DynamoDB stream and the time the connector processed it. A higher value indicates a greater lag.
#### NOTE
For shards that are stale (that is, they do not have records written to them continuously), the `MillisBehindLatest` value reported by the AWS DynamoDB Streams API’s GetRecords() operation can be high. A shard may report a large lag for up to four hours.
This is because DynamoDB Stream shards are open for writes for about four hours. The connector creates a shard consumer and continues to consume records until it sees that the next iterator for the shard is null, which indicates the shard has been closed for writes. The connector will continue to try consuming for approximately four hours and will emit the `MillisBehindLatest` value reported by the AWS DynamoDB Streams API.
For more information refer to AWS Documentation: [https://repost.aws/questions/QUA9HrpwVtRgScDcqxvNJ8pg/clarification-needed-getiteratorage-behavior-in-kinesis-data-streams](https://repost.aws/questions/QUA9HrpwVtRgScDcqxvNJ8pg/clarification-needed-getiteratorage-behavior-in-kinesis-data-streams)
In these cases, you can refer to `GetRecords.IteratorAgeMilliseconds` in AWS CloudWatch to check the lag for each shard consumer if needed.
### Monitoring checkpoint table
#### What kind of monitoring should be implemented on the checkpoint table?
You should monitor the consistent write and read throughput of the checkpoint table using
the metrics provided by AWS DynamoDB.
### RCU/WCU considerations
#### What is the RCU and write capacity unit (WCU) requirement based on the number of tables being handled by the task?
During the snapshot phase, it’s recommended to:
- Start with a high RCU on the DynamoDB tables.
- Track RCU usage on each table and tune it to an optimal level.
- For the CDC phase, RCU on the source tables is not directly relevant to the connector’s operation.
### Shared checkpoint table
#### Can multiple connectors use the same checkpoint table?
No, each unique combination of a connector and a DynamoDB table stream creates its own dedicated
checkpointing table. This ensures data consistency and prevents interference between different connectors.
#### What are the drawbacks or impacts of low RCU/WCU, does the connector handle retries or failure on DynamoDB?
Low RCU on the DynamoDB tables will lead to throttling and slow down the snapshot process. The connector does
handle retries for scan operations if it encounters throttling or RCU issues. Each scan operation can be retried
for up to 10 minutes. If the retries are exhausted within this timeframe, the task will fail.
### Checkpoint table deletion impact
#### What is the impact of the checkpoint table being deleted?
If the checkpoint table is deleted, the connector loses its progress information. As a result,
the CDC phase will start from the beginning of the DynamoDB stream.
#### Should backup be enabled for this?
Yes, it is highly recommended to enable DynamoDB table delete protection using the deletion protection property
on your checkpoint tables to prevent accidental deletion.
### Checkpoint table data size
#### What is the estimated data size for the checkpoint table?
The estimated data size for the checkpoint table is determined by the number of shards in the
DynamoDB stream being processed. Each shard in the stream corresponds to one row in the checkpoint
table.
### Reusing checkpoint table
#### Can an existing checkpoint table be reused to start from the last checkpoint?
You can use [custom offsets](#cc-amazon-dynamodb-cdc-source-offset-guidance) based on specific timestamps
to instruct the connector where to begin the CDC process.
### Connector interaction with DynamoDB
#### How does the CDC connector interact with DynamoDB, particularly during the first-time snapshot and subsequent streaming?
**Snapshot Phase**:
The connector initiates a full table scan using the [Scan API](https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_Scan.html)
provided by AWS DynamoDB. The scan operation scans the entire table to capture its current state at a specific
point in time in a paginated manner. To improve performance, the connector uses parallel scanning.
**CDC Phase**:
Once the snapshot is complete, the connector transitions to the CDC phase, which uses [Streams API](https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_Operations_Amazon_DynamoDB_Streams.html)
provided by AWS DynamoDB. The connector records a timestamp at the start of the snapshot that serves as the initial point
for the CDC phase, from which the connector begins capturing and processing changes from the stream.
### Handling shard changes and data consistency
#### How the connector ensures that all data is captured without loss during shard transitions?
Each row in the checkpointing table corresponds to a single shard in the DynamoDB stream.
The checkpointing table is unique to each `-` pair.
- Shard tracking: As the connector processes data from each shard, it updates the corresponding
row in the checkpointing table with the shard’s current offset.
- Shard addition: When new shards are added to a DynamoDB stream, the connector automatically adds
corresponding new rows to the checkpointing table. This ensures that the connector seamlessly
processes data from the newly added shards without any manual intervention or data loss.
- Shard removal: If shards are removed, the existing rows in the checkpointing table
for those shards become inactive but are generally kept for historical reference. The connector will
no longer process data associated with the removed shards.
- Parent-child relationship: The connector respects the parent-child relationships
between shards in a DynamoDB stream. It ensures that all data from a parent shard is fully processed
before it starts processing data from its child shards. This sequential processing is crucial for
maintaining the data integrity.
### DynamoDB streams consumer limit
#### What is the consumer limit for a DynamoDB stream?
Amazon DynamoDB imposes a limit of a maximum of two consumers per DynamoDB stream. For more information,
see [DynamoDB Service Quotas](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ServiceQuotas.html#limits-dynamodb-streams).
This limitation is important when planning the architecture and understanding the interactions between
different components (for example, if something goes wrong between DynamoDB and Kafka). The connector
always creates one consumer per shard in the DynamoDB stream. This leaves one remaining slot
available for other consumers.
#### What if I need to run more than one additional consumer on the same stream?
If you need to run more than one additional consumer on
the same stream, you must stop the running DynamoDB CDC Source connector first, because it is already
using one consumer per shard. Running multiple connectors on the same table will cause lease management conflicts,
stream consumption errors, and unpredictable behavior.
### IAM permissions and authentication
#### Why does my connector fail with IAM permission errors?
The connector requires specific Amazon IAM permissions to access DynamoDB tables and streams. Common permission-related
errors include:
* **Unable to build DynamoDB client**: Missing basic IAM permissions or incorrect credentials.
* **AccessDeniedException**: Missing required permissions for specific DynamoDB operations.
* **Checkpoint table creation failures**: Missing `dynamodb:CreateTable` permission.
* **CDC phase failures**: Missing `dynamodb:DeleteItem`, `dynamodb:Query`, or other checkpoint table permissions.
Verify your IAM policy includes all required permissions as documented in [IAM policy DynamoDB CDC](#cc-dynamodb-cdc-policy). The connector requires:
* **Source table permissions**: `DescribeTable`, `Scan`, `DescribeStream`, `GetRecords`, `GetShardIterator`,
`ListTagsOfResource`, `DescribeLimits`
* **Checkpoint table permissions**: `CreateTable`, `DescribeTable`, `GetItem`, `PutItem`, `Scan`, `UpdateItem`,
`DeleteItem`, `Query` on tables matching `arn:aws:dynamodb:*:*:table/-*`
* **Stream permissions**: `ListStreams`, `ListTables`, `ListGlobalTables`
* **Tag permissions**: `tag:GetResources`
When you use provider integration with IAM roles for cross-account access, ensure the trust policy includes
`sts:AssumeRole`, `sts:TagSession`, and `sts:SetSourceIdentity` actions.
#### How do I configure IAM role-based authentication instead of access keys?
The connector supports Provider Integration for IAM role-based authentication, which is recommended for enhanced
security:
1. Create a Provider Integration resource in Confluent Cloud with your Amazon IAM role configuration.
2. Configure the connector to use the provider integration instead of `aws.access.key.id` and `aws.secret.access.key`.
3. Ensure your IAM role has all required permissions documented in [IAM policy DynamoDB CDC](#cc-dynamodb-cdc-policy).
Use Provider Integration for enhanced security, cross-account access, and compliance requirements.
Provider Integration does not require static credentials. For detailed setup instructions, see [IAM roles authentication](#cc-amazon-dynamodb-cdc-source-setup-connection).
### Provisioning and configuration issues
#### Why is my connector stuck in `PROVISIONING` state?
If your connector remains in `PROVISIONING` status for extended periods with no visible errors, common causes include:
* **Invalid Avro Schema**: When you use `document.schema.source: "user_provided"`, the connector validates schemas during
provisioning.
Common causes and solutions:
* **Unsupported ENUM type**: DynamoDB connector does not support Avro ENUM types. Use `string` instead.
* **Schema redefinition**: Named record types must be defined once and referenced thereafter. Ensure record types are
defined once and referenced by name for subsequent uses.
* **Network connectivity issues**: Verify the connector can reach your DynamoDB instance. Check firewall rules, security
groups, and network policies.
Check the connector logs in the Confluent Cloud Console for schema validation errors or connection issues. If you use custom
schemas, replace ENUM types with string types and ensure proper record type definitions.
#### Why does my connector fail with `Unable to build the DynamoDB client`?
This error occurs when the connector cannot establish a connection to Amazon DynamoDB.
Common causes and solutions:
* **IAM credential issues**: Verify your `aws.access.key.id` and `aws.secret.access.key` are correct, or ensure provider
integration is correctly configured.
* **Network connectivity**: Ensure the connector can reach the DynamoDB service endpoint. For VPC-peered or PrivateLink
deployments, verify your network configuration.
### Data processing and quality issues
#### Why are some attributes missing from my records?
If you experience missing attributes from your DynamoDB records, contact Confluent [Support](https://support.confluent.io/) immediately. Missing attributes indicate a connector issue that requires investigation.
#### How does the connector handle schema for DynamoDB’s schemaless data?
The connector supports three schema modes via the `document.content.format` configuration:
* **json_string**: No schema inference. Data is stored as-is. This is the default mode.
* **schema with document.schema.source=auto_infer**: The connector infers schema from data automatically.
* **schema with document.schema.source=user_provided**: You provide explicit schema via `document.schema.value`
configuration.
Important considerations:
* Auto-inference with highly variable data can exceed Confluent Cloud Schema Registry limits and cause performance issues.
* For structured data with consistent schemas, use user-provided schemas for better performance and validation.
* Schema validation catches data quality issues early but may cause connector failures for non-conforming records.
If you are unsure what your schema looks like, temporarily start the connector in `auto_infer` mode with
Avro output format. After processing a few messages, retrieve the inferred schema from Confluent Cloud Schema Registry and use it as
the schema value when restarting the connector in `user_provided` mode.
For more information, see the Schema validation section in the connector configuration documentation.
#### Which timestamp field should I use to order DynamoDB change events?
For accurate event ordering per DynamoDB item, use the combination of `source.shard_id` and `source.seq_no`
rather than relying solely on timestamp fields.
The DynamoDB CDC connector provides several timestamp fields:
* **ts_ms**: Approximate time from DynamoDB Streams.
* **source.ts_ms**: Time when the connector processed the event.
To maintain the exact order that operations occurred in DynamoDB:
* **Primary ordering**: Use `source.seq_no` for each `source.shard_id`.
* The connector automatically handles parent-child shard relationships to maintain ordering guarantees.
* Timestamp-based ordering may not perfectly reflect DynamoDB operation order.
### Checkpoint table issues
#### Why am I getting `Can't update checkpoint - instance doesn't hold the lease` errors?
This error occurs when the checkpoint table experiences throttling due to insufficient provisioned throughput.
The connector updates the checkpoint table frequently during CDC processing.
To resolve this issue:
* Monitor the checkpoint table’s read and write capacity in the Amazon DynamoDB console.
* Increase the provisioned RCU and WCU for the checkpoint table, or switch the checkpoint table to On-Demand billing
mode for automatic scaling.
* Check the checkpoint table name pattern: `-`.
* Each connector-table combination creates its own checkpoint table, and high shard counts increase checkpoint table
operations.
If throttling persists, consider reducing the polling frequency or increasing checkpoint table capacity.
### Connector restart and recovery
#### What happens if my connector fails or restarts during the snapshot phase?
Important snapshot phase behaviors:
* **Task failure during snapshot**: The recovery is per-segment, not per-table. If you have 4 segments and 3 complete before failure, only the 1 incomplete segment will be re-executed on restart.
* **Configuration updates during snapshot**: Any configuration change restarts the snapshot process.
* **Custom offsets during snapshot**: [Custom offsets](#cc-amazon-dynamodb-cdc-source-offset-guidance) apply
to the CDC phase only, not during the initial snapshot.
* **Offsets produced after snapshot**: The connector only produces Kafka offsets after completing the snapshot phase,
which signals transition to CDC mode.
To minimize snapshot restart risk:
* Avoid connector configuration changes during the snapshot phase.
* Ensure adequate RCU provisioning for fast snapshot completion.
* Use one task per table during the snapshot phase.
## Next steps
* For an example that shows fully managed Confluent Cloud connectors in action with
Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html).
This example also shows how to use Confluent CLI to manage your resources in
Confluent Cloud.
[](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)
* Try [Confluent Cloud on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-g5ujul6iovvcy?trk=14575e70-1766-4f20-8083-0c2757a1ec75&sc_channel=el)
with $1000 of free usage for 30 days, and pay as you go. No credit card is
required.