Crate go_zoom_kinesis

Go Zoom Kinesis - A robust AWS Kinesis stream processor

This library provides a production-ready implementation for processing AWS Kinesis streams with features like:

• Automatic checkpointing with pluggable storage backends
• Configurable retry logic with exponential backoff
• Concurrent shard processing with rate limiting
• Comprehensive monitoring and metrics
• Graceful shutdown handling

Basic Usage

use go_zoom_kinesis::{
    KinesisProcessor, ProcessorConfig, RecordProcessor,
    processor::RecordMetadata, processor::InitialPosition,
    store::InMemoryCheckpointStore,
    monitoring::MonitoringConfig,
    error::{ProcessorError, ProcessingError},
};
use aws_sdk_kinesis::{Client, types::Record};
use std::time::Duration;
use async_trait::async_trait;

#[derive(Clone)]
struct MyProcessor;

#[async_trait]
impl RecordProcessor for MyProcessor {
    type Item = ();

    async fn process_record<'a>(
        &self,
        record: &'a Record,
        metadata: RecordMetadata<'a>,
    ) -> Result<Option<Self::Item>, ProcessingError> {
        println!("Processing record: {:?}", record);
        Ok(None)
    }
}

#[tokio::main]
async fn main() -> Result<(), ProcessorError> {
    let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
    let client = Client::new(&config);

    let config = ProcessorConfig {
        stream_name: "my-stream".to_string(),
        batch_size: 100,
        api_timeout: Duration::from_secs(30),
        processing_timeout: Duration::from_secs(300),
        max_retries: Some(3),
        shard_refresh_interval: Duration::from_secs(60),
        initial_position: InitialPosition::TrimHorizon,
        prefer_stored_checkpoint: true,
        monitoring: MonitoringConfig {
            enabled: true,
            ..Default::default()
        },
        ..Default::default()
    };

    let processor = MyProcessor;
    let store = InMemoryCheckpointStore::new();

    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
    let (processor, _monitoring_rx) = KinesisProcessor::new(
        config,
        processor,
        client,
        store,
    );

    processor.run(shutdown_rx).await
}

Stream Position Configuration

use go_zoom_kinesis::{RecordProcessor, error::ProcessingError};
use go_zoom_kinesis::processor::RecordMetadata;
use aws_sdk_kinesis::types::Record;
use async_trait::async_trait;
use anyhow::Result;

struct MyProcessor;

#[async_trait]
impl RecordProcessor for MyProcessor {
    // Define the associated type - in this case we don't produce any items
    type Item = ();

    async fn process_record<'a>(
        &self,
        record: &'a Record,
        metadata: RecordMetadata<'a>,
   ) -> Result<Option<Self::Item>, ProcessingError> {
       match process_data(record).await {
            Ok(_) => Ok(None),  // No item produced
            Err(e) => {
                // Custom error handling with metadata context
                tracing::error!(
                    error = %e,
                    shard_id = %metadata.shard_id(),
                    attempt = %metadata.attempt_number(),
                    "Failed to process record"
                );
                Err(ProcessingError::soft(e)) // Will be retried
            }
        }
    }
}

// Example processing function
async fn process_data(_record: &Record) -> Result<()> {
   Ok(())
}

DynamoDB Checkpoint Store

use go_zoom_kinesis::store::DynamoDbCheckpointStore;


async fn example() -> anyhow::Result<()> {
    let config = aws_config::load_from_env().await;
    let dynamo_client = aws_sdk_dynamodb::Client::new(&config);
    let checkpoint_store = DynamoDbCheckpointStore::new(
        dynamo_client,
        "checkpoints-table".to_string(),
        "my-app-".to_string(),
    );
    Ok(())
}

Re-exports

pub use error::ProcessorError;

pub use error::Result;

pub use processor::KinesisProcessor;

pub use processor::ProcessorConfig;

pub use retry::Backoff;

pub use retry::ExponentialBackoff;

pub use crate::processor::RecordProcessor;

pub use crate::store::CheckpointStore;

pub use crate::store::memory::InMemoryCheckpointStore;

pub use crate::store::dynamodb::DynamoDbCheckpointStore;

Modules

client

error

Error types for the Kinesis processor

monitoring

Monitoring system for tracking Kinesis processor performance and health

processor

Core processor implementation for handling Kinesis streams

retry

Retry and backoff functionality for the Kinesis processor

store

Checkpoint storage implementations for the Kinesis processor