feat(connectors): add DynamoDB sink connector - #3996
Conversation
Writes each message as an item, in batches of 25, which is the BatchWriteItem limit. Closes apache#3731
A redelivered message overwrites the same item. Items sharing a key within one batch are dropped, because BatchWriteItem rejects them.
An oversized item or an invalid key type fails the whole batch, so the message is dropped and logged instead.
Throttling, server errors and the items BatchWriteItem leaves unprocessed are retried with exponential backoff.
A key field that does not match the table now fails while the sink opens, instead of on the first write.
They run against DynamoDB Local and cover batch splitting, a table with a sort key, and rewriting the same message ids.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3996 +/- ##
============================================
+ Coverage 84.76% 84.82% +0.06%
Complexity 1405 1405
============================================
Files 1224 1225 +1
Lines 177971 178873 +902
Branches 144285 145187 +902
============================================
+ Hits 150854 151732 +878
+ Misses 23100 23098 -2
- Partials 4017 4043 +26
🚀 New features to boost your workflow:
|
|
/request-review @hubcio |
slbotbm
left a comment
There was a problem hiding this comment.
In addition to the comments, you have not added the dynamodb connector to the CI, so while the plugin compiles and the unit tests run, integration::connectors::dynamodb do not run if changes in dynamodb connector are detected.
| item.entry(self.partition_key_field.clone()) | ||
| .or_insert_with(|| { | ||
| AttributeValue::S(build_message_key( | ||
| topic_metadata, | ||
| messages_metadata, | ||
| message.id, | ||
| )) | ||
| }); |
There was a problem hiding this comment.
DynamoDbSink::build_item derives the default iggy_id from build_message_key, which uses message.id. IggyMessage::new defaults an omitted message ID to 0, so ordinary messages built without .id(...) receive the same default DynamoDB key within a stream partition. DynamoDbSink::deduplicate_items then keeps only the newest item in a consume call, and later batches overwrite the same DynamoDB record. The integration helper always sets an ID, so it does not cover the default. Use the stable, unique message offset in the generated key, or require and validate a non-default message ID.
| fn build_message_key( | ||
| topic_metadata: &TopicMetadata, | ||
| messages_metadata: &MessagesMetadata, | ||
| message_id: u128, | ||
| ) -> String { | ||
| format!( | ||
| "{}:{}:{}:{message_id}", | ||
| topic_metadata.stream, topic_metadata.topic, messages_metadata.partition_id | ||
| ) | ||
| } |
There was a problem hiding this comment.
build_message_key serializes stream, topic, partition, and message ID with an unescaped : separator. Stream and topic names permit :, so (stream = "a:b", topic = "c") and (stream = "a", topic = "b:c") generate the same key for the same partition and message ID. A connector configured for both topics can overwrite unrelated items. Use length-delimited or binary encoding, or a collision-resistant digest, and add collision tests.
| fn key_signature(&self, item: &HashMap<String, AttributeValue>) -> String { | ||
| let mut signature = key_attribute_signature(item.get(&self.partition_key_field)); | ||
| if let Some(sort_key_field) = &self.sort_key_field { | ||
| signature.push('|'); | ||
| signature.push_str(&key_attribute_signature(item.get(sort_key_field))); | ||
| } | ||
| signature |
There was a problem hiding this comment.
DynamoDbSink::deduplicate_items has an independent collision bug for composite String keys. key_signature joins unescaped S:/N: values with |, so {pk: S("a"), sk: S("b|S:c")} collides with {pk: S("a|S:b"), sk: S("c")}. The second item silently replaces the first before the request is sent. Use a typed, length-delimited key representation rather than a display string.
| self.validate_key_schema( | ||
| description | ||
| .table | ||
| .and_then(|table| table.key_schema) | ||
| .unwrap_or_default(), | ||
| )?; |
There was a problem hiding this comment.
DynamoDbSink::open passes only TableDescription::key_schema to DynamoDbSink::validate_key_schema. validate_key_attribute consequently accepts any non-empty String, Number, or Binary value and never enforces the 2,048-byte partition-key or 1,024-byte sort-key limits. DynamoDB requires an exact key type: a table with an N/B partition key still opens, although the connector's generated partition key is always S; a missing sort key is always generated as N. A mismatched or overlong key causes DynamoDB to reject the whole BatchWriteItem request. Retain the table's attribute definitions, validate each key's exact type and byte length before batching, and reject configurations incompatible with generated key values.
| fn estimate_item_size(item: &HashMap<String, AttributeValue>) -> usize { | ||
| item.iter() | ||
| .map(|(name, value)| name.len() + attribute_value_size(value)) | ||
| .sum() | ||
| } | ||
|
|
||
| fn attribute_value_size(value: &AttributeValue) -> usize { | ||
| match value { | ||
| AttributeValue::S(text) => text.len(), | ||
| AttributeValue::N(number) => number.len(), | ||
| AttributeValue::B(blob) => blob.as_ref().len(), | ||
| AttributeValue::Bool(_) | AttributeValue::Null(_) => 1, | ||
| AttributeValue::Ss(values) => values.iter().map(String::len).sum(), | ||
| AttributeValue::Ns(values) => values.iter().map(String::len).sum(), | ||
| AttributeValue::Bs(values) => values.iter().map(|blob| blob.as_ref().len()).sum(), | ||
| AttributeValue::L(values) => values.iter().map(attribute_value_size).sum(), | ||
| AttributeValue::M(values) => estimate_item_size(values), | ||
| _ => 0, | ||
| } |
There was a problem hiding this comment.
estimate_item_size and attribute_value_size are not DynamoDB item-size calculations. They omit the 100-byte base item overhead and Map/List overhead, and size Numbers by their string representation rather than DynamoDB's number-sizing rules. Nested items can therefore pass the max_item_size check but make DynamoDB reject the entire request, contrary to the README's promise that oversized records are skipped. Implement an exact or deliberately conservative size check and add nested boundary tests. DynamoDB documents the relevant overhead in https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/CapacityUnitCalculations.html .
| fn is_transient_code(code: &str) -> bool { | ||
| matches!( | ||
| code, | ||
| "ProvisionedThroughputExceededException" | ||
| | "RequestLimitExceeded" | ||
| | "ThrottlingException" | ||
| | "InternalServerError" | ||
| | "ServiceUnavailable" | ||
| | "TransactionInProgressException" | ||
| ) | ||
| } |
There was a problem hiding this comment.
is_transient_code treats ReplicatedWriteConflictException as permanent. DynamoDB marks this multi-Region strongly consistent global-table conflict retryable, but the connector returns a permanent error rather than applying its configured backoff. Add the error code and a regression test.
| if self.config.access_key_id.is_some() != self.config.secret_access_key.is_some() { | ||
| return Err(Error::InvalidConfigValue( | ||
| "Partially configured credentials. You must provide both access_key_id \ | ||
| and secret_access_key, or omit both." | ||
| .to_owned(), | ||
| )); | ||
| } | ||
|
|
||
| let mut loader = aws_config::defaults(BehaviorVersion::latest()); |
There was a problem hiding this comment.
DynamoDbSink::build_client only checks whether access_key_id and secret_access_key are paired. A configured session_token without that pair is silently ignored and the default AWS credential chain is used. Reject that invalid configuration so an operator cannot believe STS credentials are active when they are not.
| if let Some(endpoint) = &self.config.endpoint { | ||
| info!("Using custom DynamoDB endpoint: {endpoint}"); | ||
| loader = loader.endpoint_url(endpoint); | ||
| } |
There was a problem hiding this comment.
build_client logs custom endpoint URLs verbatim. An arbitrary URL may contain user information or query credentials. Redact the URL at the log site, consistent with connector secret-handling guidance.
| 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. |
There was a problem hiding this comment.
This section says key fields are checked against the table on startup, but the implementation checks only names and HASH/RANGE roles. It should document exact S/N/B compatibility, the generated partition key's S requirement, and the generated sort key's N requirement. Its statement that a payload key always wins also conflicts with DynamoDbSink::build_item, where enabled Iggy metadata overwrites a colliding payload field before generated keys are filled.
| ## 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. |
There was a problem hiding this comment.
DynamoDbSink::build_item never consumes ConsumedMessage::headers. The README should explicitly state that user headers are not persisted, or the connector should provide a documented representation for them.
Which issue does this PR address?
Closes #3731
Rationale
Iggy can write to several databases, but not to Amazon DynamoDB, so users have to write their own consumer for it.
What changed?
Stream messages could not reach a DynamoDB table without custom code. This adds a sink connector that maps each message to a DynamoDB item and writes it with
BatchWriteItem, splitting batches at the 25 item limit and retrying throttled requests and unprocessed items with exponential backoff. The item key is built from the stream, topic, partition and message id when the payload does not carry the configured key field, so a redelivered message overwrites the same item. Records that DynamoDB would reject, such as items over 400 KB or an invalid key type, are skipped with a warning instead of failing the whole batch.The sink uses the official
aws-sdk-dynamodb, so the repo now carries the AWS SDK next to therust-s3stack that the S3 sink uses. DynamoDB has no comparable third-party client.The commits are split so that each one is a working increment: base sink, keys, skipped records, retries, key schema check, then tests and docs.
Local Execution
AI Usage