feat(connectors): add DynamoDB sink connector

Cargo.toml

@@ -36,6 +36,7 @@ members = [
     "core/connectors/sinks/clickhouse_sink",
     "core/connectors/sinks/delta_sink",
     "core/connectors/sinks/doris_sink",
+    "core/connectors/sinks/dynamodb_sink",
     "core/connectors/sinks/elasticsearch_sink",
     "core/connectors/sinks/http_sink",
     "core/connectors/sinks/iceberg_sink",
@@ -105,6 +106,8 @@ async-channel = "2.5.0"
 async-dropper = { version = "0.3.1", features = ["tokio", "simple"] }
 async-trait = "0.1.91"
 async_zip = { version = "0.0.18", features = ["tokio", "lzma", "bzip2", "xz", "deflate", "zstd"] }
+aws-config = { version = "1.9.0", features = ["behavior-version-latest"] }
+aws-sdk-dynamodb = "1.117.0"
 axum = { version = "0.8.9", features = ["macros"] }
 axum-server = { version = "0.8.0", features = ["tls-rustls"] }
 base64 = "0.22.1"

core/connectors/README.md

@@ -81,6 +81,7 @@ Each sink should have its own, custom configuration, which is passed along with
 ### Available Sinks
 
 - **Doris Sink** - loads JSON messages into Apache Doris tables via the Stream Load HTTP API
+- **DynamoDB Sink** - writes messages to Amazon DynamoDB tables
 - **Elasticsearch Sink** - sends messages to Elasticsearch indices
 - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
 - **Meilisearch Sink** - indexes messages in Meilisearch

core/connectors/sinks/README.md

@@ -9,6 +9,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 | Sink | Description |
 | ---- | ----------- |
 | **doris_sink** | Loads JSON messages into Apache Doris tables via the Stream Load HTTP API |
+| **dynamodb_sink** | Writes messages to Amazon DynamoDB tables with batched, key-deterministic writes |
 | **elasticsearch_sink** | Sends messages to Elasticsearch indices for full-text search and analytics |
 | **iceberg_sink** | Writes data to Apache Iceberg tables via REST catalog with S3/GCS/Azure storage |
 | **influxdb_sink** | Writes messages to InfluxDB as line-protocol points; supports both V2 (org/bucket, Flux) and V3 (db, SQL) |

core/connectors/sinks/dynamodb_sink/Cargo.toml

@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_dynamodb_sink"
+version = "0.5.0-edge.4"
+description = "Iggy DynamoDB sink connector for writing stream messages to Amazon DynamoDB tables"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "dynamodb", "sink"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+aws-config = { workspace = true }
+aws-sdk-dynamodb = { workspace = true }
+humantime = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+simd-json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[lints]
+workspace = true

core/connectors/sinks/dynamodb_sink/README.md

@@ -0,0 +1,95 @@
+# DynamoDB Sink Connector
+
+A sink connector that consumes messages from Iggy streams and writes them to an
+Amazon DynamoDB table with `BatchWriteItem` through the official AWS SDK.
+
+## Configuration
+
+```toml
+[plugin_config]
+table = "iggy_messages"
+region = "us-east-1"
+# endpoint = "http://localhost:8000"
+# access_key_id = "..."
+# secret_access_key = "..."
+# session_token = "..."
+partition_key_field = "iggy_id"
+# sort_key_field = "iggy_offset"
+batch_size = 25
+include_metadata = true
+include_checksum = true
+include_origin_timestamp = true
+max_item_size = 409600
+max_retries = 3
+retry_delay = "500ms"
+max_retry_delay = "5s"
+verbose_logging = false
+```
+
+- `table`: Target DynamoDB table. The table must already exist.
+- `region`: AWS region. Falls back to the default AWS region chain when unset.
+- `endpoint`: Custom endpoint URL, for DynamoDB Local or a VPC endpoint.
+- `access_key_id` / `secret_access_key` / `session_token`: Static credentials.
+  Provide both `access_key_id` and `secret_access_key`, or neither. When both
+  are omitted the connector uses the default AWS credential chain.
+- `partition_key_field`: Item attribute used as the table partition key.
+  Defaults to `iggy_id`.
+- `sort_key_field`: Item attribute used as the table sort key. Only set this
+  when the table has a sort key.
+- `batch_size`: Items per `BatchWriteItem` request. Defaults to `25`, which is
+  also the DynamoDB limit, so larger values are clamped.
+- `include_metadata`: Add `iggy_stream`, `iggy_topic`, `iggy_partition_id`,
+  `iggy_offset`, and `iggy_timestamp` to each item. Defaults to `true`.
+- `include_checksum`: Add `iggy_checksum`. Defaults to `true`.
+- `include_origin_timestamp`: Add `iggy_origin_timestamp`. Defaults to `true`.
+- `max_item_size`: Maximum item size in bytes. Defaults to `409600` (400 KB),
+  which is also the DynamoDB limit, so larger values are clamped.
+- `max_retries`: Retries after the first attempt. Defaults to `3`.
+- `retry_delay`: First retry delay as a humantime string. Defaults to `500ms`.
+- `max_retry_delay`: Upper bound of the exponential backoff. Defaults to `5s`.
+- `verbose_logging`: Log per-batch results at info level. Defaults to `false`.
+
+## Behavior
+
+JSON objects are written attribute by attribute, so a message field becomes a
+DynamoDB attribute of the matching type. JSON arrays and scalars are nested
+under a `payload` attribute, because a DynamoDB item must be a map. Text
+payloads go into `payload` as a string. Raw payloads are parsed as JSON when
+possible, otherwise they are stored as binary. Protobuf, FlatBuffer, and Avro
+payloads are not supported and are skipped with a warning.
+
+Metadata attributes are written after the payload, so they overwrite payload
+fields of the same name.
+
+## Keys and Idempotency
+
+`BatchWriteItem` uses `PutRequest`, which overwrites an existing item with the
+same primary key. Redelivery of the same message therefore writes the same item
+again as long as the key is deterministic.
+
+When the payload does not carry the configured `partition_key_field`, the
+connector injects a key built from the stream, topic, partition, and message ID.
+When `sort_key_field` is configured and missing, the message offset is injected.
+A payload value always wins over the injected one, so a message that carries the
+key field with an empty value, or with a value that is neither a string, a
+number, nor binary, is skipped rather than falling back to the injected key,
+because DynamoDB would reject the whole batch.
+
+The key fields are checked against the table on startup. A `partition_key_field`
+or `sort_key_field` that does not match the table key schema fails the connector
+while it opens, instead of on the first write.
+
+DynamoDB rejects a request that writes the same key twice, so within one
+`consume()` call only the newest item per key is sent.
+
+## Retries
+
+Unprocessed items returned by `BatchWriteItem` are retried with exponential
+backoff, so a partially throttled batch is not silently dropped. Throttling and
+server errors such as `ProvisionedThroughputExceededException`,
+`ThrottlingException`, and `InternalServerError` are retried the same way, which
+covers both provisioned and on-demand capacity modes. Validation and access
+errors are permanent and returned without a retry.
+
+Items larger than `max_item_size` are logged and skipped instead of failing the
+whole batch.

core/connectors/sinks/dynamodb_sink/config.toml

@@ -0,0 +1,51 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+type = "sink"
+key = "dynamodb"
+enabled = true
+version = 0
+name = "DynamoDB sink"
+path = "../../target/release/libiggy_connector_dynamodb_sink"
+verbose = false
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "dynamodb_sink"
+
+[plugin_config]
+table = "iggy_messages"
+region = "us-east-1"
+# endpoint = "http://localhost:8000"     # uncomment for DynamoDB Local
+# access_key_id = "..."                  # omit to use the default AWS credential chain
+# secret_access_key = "..."              # omit to use the default AWS credential chain
+# session_token = "..."                  # only needed for temporary credentials
+partition_key_field = "iggy_id"
+# sort_key_field = "iggy_offset"
+batch_size = 25
+include_metadata = true
+include_checksum = true
+include_origin_timestamp = true
+max_item_size = 409600
+max_retries = 3
+retry_delay = "500ms"
+max_retry_delay = "5s"
+verbose_logging = false