Consuming Kinesis event with Lambda using Rust with batch item handling.

Author: kazhura-aws

integration-kinesis-to-lambda-with-batch-item-handling/main.rs

@@ -0,0 +1,74 @@
+use aws_lambda_events::{
+    event::kinesis::KinesisEvent,
+    kinesis::KinesisEventRecord,
+    streams::{KinesisBatchItemFailure, KinesisEventResponse},
+};
+use lambda_runtime::{run, service_fn, Error, LambdaEvent};
+
+static UNDEFINED: &str = "undefined";
+
+async fn function_handler(event: LambdaEvent<KinesisEvent>) -> Result<KinesisEventResponse, Error> {
+    let mut response = KinesisEventResponse {
+        batch_item_failures: vec![],
+    };
+
+    if event.payload.records.is_empty() {
+        tracing::info!("No records found. Exiting.");
+        return Ok(response);
+    }
+
+    for record in &event.payload.records {
+        tracing::info!(
+            "EventId: {}",
+            record.event_id.as_deref().unwrap_or(UNDEFINED)
+        );
+
+        let record_processing_result = process_record(record);
+
+        if record_processing_result.is_err() {
+            response.batch_item_failures.push(KinesisBatchItemFailure {
+                item_identifier: record.kinesis.sequence_number.clone(),
+            });
+            /* Since we are working with streams, we can return the failed item immediately.
+            Lambda will immediately begin to retry processing from this failed item onwards. */
+            return Ok(response);
+        }
+    }
+
+    tracing::info!(
+        "Successfully processed {} records",
+        event.payload.records.len()
+    );
+
+    Ok(response)
+}
+
+fn process_record(record: &KinesisEventRecord) -> Result<(), Error> {
+    let record_data = std::str::from_utf8(record.kinesis.data.as_slice());
+
+    if record_data.is_err() {
+        let err = record_data.err().unwrap();
+        tracing::error!("Error: {}", err);
+        return Err(Error::from(err));
+    }
+
+    let record_data = record_data.unwrap();
+
+    // do something interesting with the data
+    tracing::info!("Data: {}", record_data);
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Error> {
+    tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::INFO)
+        // disable printing the name of the module in every log line.
+        .with_target(false)
+        // disabling time is handy because CloudWatch will add the ingestion time.
+        .without_time()
+        .init();
+
+    run(service_fn(function_handler)).await
+}

integration-kinesis-to-lambda-with-batch-item-handling/snippet-data.json

@@ -74,6 +74,17 @@
               "language": "dotnet"
             }
           ]
+        },
+        {
+          "id": "Rust",
+          "title": "Usage Example with Rust:",
+          "description": "Consuming Kinesis event with Lambda using Rust with batch item handling.",
+          "snippets": [
+            {
+              "snippetPath": "main.rs",
+              "language": "rust"
+            }
+          ]
         }
       ]
     }