diff --git a/lambda-events/src/event/kafka/mod.rs b/lambda-events/src/event/kafka/mod.rs index 5ca28b3d..187c58e6 100644 --- a/lambda-events/src/event/kafka/mod.rs +++ b/lambda-events/src/event/kafka/mod.rs @@ -55,6 +55,90 @@ pub struct KafkaRecord { pub other: serde_json::Map, } +/// `KafkaEventResponse` is the outer structure to report batch item failures for `KafkaEvent`. +#[non_exhaustive] +#[cfg_attr(feature = "builders", derive(Builder))] +#[derive(Debug, Default, Clone, Eq, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct KafkaEventResponse { + pub batch_item_failures: Vec, + /// Catchall to catch any additional fields that were present but not explicitly defined by this struct. + /// Enabled with Cargo feature `catch-all-fields`. + /// If `catch-all-fields` is disabled, any additional fields that are present will be ignored. + #[cfg(feature = "catch-all-fields")] + #[cfg_attr(docsrs, doc(cfg(feature = "catch-all-fields")))] + #[serde(flatten)] + #[cfg_attr(feature = "builders", builder(default))] + pub other: serde_json::Map, +} + +impl KafkaEventResponse { + /// Add a failed Kafka item identifier to the batch response. + /// + /// Lambda retries the identified records when `ReportBatchItemFailures` is enabled on the + /// Kafka event source mapping. Returning an error from the handler still retries the whole + /// batch. + pub fn add_failure(&mut self, item_identifier: KafkaItemIdentifier) { + self.batch_item_failures.push(KafkaBatchItemFailure { + item_identifier, + ..Default::default() + }); + } + + /// Set all failed Kafka item identifiers in the batch response. + /// + /// This replaces any previously registered failures. + pub fn set_failures(&mut self, item_identifiers: I) + where + I: IntoIterator, + { + self.batch_item_failures = item_identifiers + .into_iter() + .map(|item_identifier| KafkaBatchItemFailure { + item_identifier, + ..Default::default() + }) + .collect(); + } +} + +/// `KafkaBatchItemFailure` is an individual Kafka record which failed processing. +#[non_exhaustive] +#[cfg_attr(feature = "builders", derive(Builder))] +#[derive(Debug, Default, Clone, Eq, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct KafkaBatchItemFailure { + pub item_identifier: KafkaItemIdentifier, + /// Catchall to catch any additional fields that were present but not explicitly defined by this struct. + /// Enabled with Cargo feature `catch-all-fields`. + /// If `catch-all-fields` is disabled, any additional fields that are present will be ignored. + #[cfg(feature = "catch-all-fields")] + #[cfg_attr(docsrs, doc(cfg(feature = "catch-all-fields")))] + #[serde(flatten)] + #[cfg_attr(feature = "builders", builder(default))] + pub other: serde_json::Map, +} + +/// `KafkaItemIdentifier` identifies a Kafka record for a partial batch response. +#[non_exhaustive] +#[cfg_attr(feature = "builders", derive(Builder))] +#[derive(Debug, Default, Clone, Eq, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct KafkaItemIdentifier { + /// The topic-partition key from the Kafka event's `records` map. + pub partition: String, + /// The Kafka record offset. + pub offset: i64, + /// Catchall to catch any additional fields that were present but not explicitly defined by this struct. + /// Enabled with Cargo feature `catch-all-fields`. + /// If `catch-all-fields` is disabled, any additional fields that are present will be ignored. + #[cfg(feature = "catch-all-fields")] + #[cfg_attr(docsrs, doc(cfg(feature = "catch-all-fields")))] + #[serde(flatten)] + #[cfg_attr(feature = "builders", builder(default))] + pub other: serde_json::Map, +} + #[cfg(test)] mod test { use super::*; @@ -68,4 +152,54 @@ mod test { let reparsed: KafkaEvent = serde_json::from_slice(output.as_bytes()).unwrap(); assert_eq!(parsed, reparsed); } + + #[test] + #[cfg(feature = "kafka")] + fn kafka_event_response_serializes_item_identifiers() { + let mut response = KafkaEventResponse::default(); + response.add_failure(KafkaItemIdentifier { + partition: String::from("some.topic-3"), + offset: 42, + ..Default::default() + }); + + let serialized = serde_json::to_value(response).unwrap(); + + assert_eq!( + serialized, + serde_json::json!({ + "batchItemFailures": [{ + "itemIdentifier": { + "partition": "some.topic-3", + "offset": 42, + } + }] + }) + ); + } + + #[test] + #[cfg(feature = "kafka")] + fn kafka_event_response_sets_failures() { + let mut response = KafkaEventResponse::default(); + response.set_failures([ + KafkaItemIdentifier { + partition: String::from("some.topic-3"), + offset: 42, + ..Default::default() + }, + KafkaItemIdentifier { + partition: String::from("some.topic-4"), + offset: 43, + ..Default::default() + }, + ]); + + assert_eq!(response.batch_item_failures.len(), 2); + assert_eq!( + response.batch_item_failures[0].item_identifier.partition, + "some.topic-3" + ); + assert_eq!(response.batch_item_failures[1].item_identifier.offset, 43); + } }