native-aws-kcl-py 0.1.1

Native AWS Kinesis Consumer Library for Python - KCL implementation with lease coordination, resharding, and metrics

native-aws-kcl-py

A native Python implementation of the AWS Kinesis Client Library (KCL) for consuming Kinesis streams. This library provides the same functionality as the Java-based KCL but is written entirely in Python with an async/await concurrency model.

Features

• Shard Discovery: Automatically discovers and tracks Kinesis shards
• Lease Management: Coordinates shard-to-worker assignment using pluggable storage (DynamoDB, Redis, S3, PostgreSQL, MySQL, MongoDB)
• Lease Renewal: Maintains heartbeats to prevent lease expiration
• Failover & Rebalancing: Automatically handles worker failures and redistributes shards
• Checkpoint Coordination: Persists processing progress with pluggable storage backends
• Graceful Shutdown: Handles shutdown signals and shard end events
• Resharding Support: Handles shard splits and merges automatically
• Enhanced Fan-Out (EFO): Optional dedicated throughput mode with 2MB/s per consumer per shard
• Exponential Backoff with Jitter: Handles throttling gracefully following AWS best practices

Installation

Base Installation (DynamoDB - Default)

pip install native-aws-kcl-py

Or with Poetry:

poetry add native-aws-kcl-py

With Optional Persistence Backends

Install with additional drivers for your chosen persistence backend:

# S3 persistence (uses aioboto3, included by default)
pip install native-aws-kcl-py

# Redis persistence
pip install native-aws-kcl-py aioredis

# PostgreSQL persistence
pip install native-aws-kcl-py asyncpg

# MySQL persistence
pip install native-aws-kcl-py aiomysql

# MongoDB persistence
pip install native-aws-kcl-py motor

Note: The persistence drivers (aioredis, asyncpg, aiomysql, motor) are optional dependencies. Only install what you need. If you try to use a persistence type without its driver, you'll get a clear error message.

Quick Start

Basic Usage

import asyncio
from native_aws_kcl_py import (
    Scheduler,
    KinesisConsumerConfig,
    AwsConfig,
    ShardRecordProcessor,
    ShardRecordProcessorFactory,
    InitializationInput,
    ProcessRecordsInput,
    LeaseLostInput,
    ShardEndedInput,
    ShutdownRequestedInput,
)


class MyRecordProcessor(ShardRecordProcessor):
    """Your record processor implementation."""

    async def initialize(self, input: InitializationInput) -> None:
        print(f"Starting to process shard {input.shard_id}")

    async def process_records(self, input: ProcessRecordsInput) -> None:
        for record in input.records:
            data = record.data.decode('utf-8')
            print(f"Processing: {data}")

        # Checkpoint after processing
        if input.records:
            await input.checkpointer.checkpoint()

    async def lease_lost(self, input: LeaseLostInput) -> None:
        print(f"Lost lease for shard {input.shard_id}")

    async def shard_ended(self, input: ShardEndedInput) -> None:
        await input.checkpointer.checkpoint()
        print(f"Shard {input.shard_id} ended")

    async def shutdown_requested(self, input: ShutdownRequestedInput) -> None:
        await input.checkpointer.checkpoint()
        print(f"Shutdown requested for shard {input.shard_id}")


class MyProcessorFactory(ShardRecordProcessorFactory):
    def create_processor(self) -> ShardRecordProcessor:
        return MyRecordProcessor()


async def main():
    config = KinesisConsumerConfig(
        aws=AwsConfig(region="eu-west-1"),
        application_name="my-consumer-app",
        stream_name="my-kinesis-stream",
        record_processor_factory=MyProcessorFactory(),
    )

    scheduler = Scheduler(config)

    # Handle signals for graceful shutdown
    import signal
    
    def signal_handler():
        asyncio.create_task(scheduler.shutdown())
    
    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGTERM, signal_handler)
    loop.add_signal_handler(signal.SIGINT, signal_handler)

    await scheduler.run()


if __name__ == "__main__":
    asyncio.run(main())

Core Concepts

Scheduler

The Scheduler (also known as Worker in KCL) is the main entry point. It coordinates all components:

• LeaseCoordinator: Manages shard leases for distributed processing
• ShardDetector: Discovers shards in the stream
• ShardSyncer: Synchronizes shards with leases
• ShardConsumers: Process records from individual shards

Record Processors

Record processors implement the ShardRecordProcessor interface to handle records from a shard. Each processor must implement the following lifecycle methods:

• initialize(): Called when a shard is assigned to the worker
• process_records(): Called to process a batch of records
• lease_lost(): Called when the lease is lost to another worker
• shard_ended(): Called when the shard reaches its end (resharding)
• shutdown_requested(): Called when the worker is shutting down

Lease Management

Leases coordinate which worker processes which shard. The library supports multiple storage backends:

• DynamoDB (default)
• Redis
• S3
• PostgreSQL
• MySQL
• MongoDB

Configuration

KinesisConsumerConfig

from native_aws_kcl_py import (
    KinesisConsumerConfig,
    AwsConfig,
    RetrievalConfig,
    CheckpointConfig,
    CoordinatorConfig,
    LifecycleConfig,
    MetricsConfig,
    InitialPositionInStream,
    InitialPositionInStreamExtended,
)

config = KinesisConsumerConfig(
    aws=AwsConfig(
        region="eu-west-1",
        kinesis_endpoint=None,  # Custom endpoint for LocalStack
        dynamodb_endpoint=None,
    ),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    worker_identifier="worker-1",  # Auto-generated if not provided
    initial_position_in_stream=InitialPositionInStreamExtended(
        initial_position_in_stream=InitialPositionInStream.TRIM_HORIZON,
    ),
    retrieval=RetrievalConfig(
        max_records=10000,
        idle_time_between_reads_millis=1000,
    ),
    checkpoint=CheckpointConfig(
        call_process_records_even_for_empty_record_list=False,
        always_start_from_latest=False,
    ),
    coordinator=CoordinatorConfig(
        lease_duration_millis=10000,
        max_leases_for_worker=10,
    ),
    lifecycle=LifecycleConfig(
        task_execution_timeout_millis=300000,
        max_consecutive_failures=10,
    ),
    metrics=MetricsConfig(
        level="NONE",  # NONE, SUMMARY, or DETAILED
    ),
)

Checkpointing

Checkpointing saves processing progress. If the worker restarts, it resumes from the last checkpoint.

Manual Checkpointing

class MyProcessor(ShardRecordProcessor):
    async def process_records(self, input: ProcessRecordsInput) -> None:
        for record in input.records:
            await self.process(record)
        
        # Checkpoint after processing
        await input.checkpointer.checkpoint()
        
        # Or checkpoint at a specific sequence number
        await input.checkpointer.checkpoint(
            sequence_number="49590...",
            sub_sequence_number=0,
        )

Always Start From Latest (Skip Checkpoints)

In some scenarios, you may want to always start processing from the latest records in the stream, ignoring any existing checkpoints. This is useful when:

• You only care about real-time data and want to skip any backlog
• You're running a stateless consumer that doesn't need to track progress
• You want to restart processing from the current position after a deployment

config = KinesisConsumerConfig(
    aws=AwsConfig(region="eu-west-1"),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    checkpoint=CheckpointConfig(
        always_start_from_latest=True,
    ),
)

Important: Lease coordination (shard assignment, failover, distributed processing) still uses persistence. Only checkpoint data reading/writing is affected.

Warning: When always_start_from_latest is enabled:

• The consumer will always start from the newest records
• Any records that arrived while the consumer was stopped will be skipped
• Existing checkpoints will be ignored (but not deleted)

Processing Empty Record Batches

By default, the process_records method in your record processor is only called when there are records to process. You can change this behavior to receive callbacks even when Kinesis returns empty batches.

Configuration

config = KinesisConsumerConfig(
    aws=AwsConfig(region="eu-west-1"),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    checkpoint=CheckpointConfig(
        call_process_records_even_for_empty_record_list=True,
    ),
)

Or via environment variable:

export KCL_CALL_PROCESS_RECORDS_EVEN_FOR_EMPTY_LIST=true

When to Use

Enable this setting when you need to:

• Implement heartbeats or periodic tasks - Execute logic at regular intervals even without new data
• Force checkpoint updates during idle periods - Ensure checkpoints are updated even when no records arrive
• Detect long idle periods - Monitor and alert when no data is received for extended periods
• Handle state transitions - React to shard lifecycle changes immediately

Behavior

Setting	Records Available	process_records Called
False (default)	Yes	Yes
False (default)	No	No
True	Yes	Yes
True	No	Yes (with empty records list)

Trade-offs

• Extra overhead: More frequent process_records invocations even during idle periods
• Code handling: Your processor must handle empty record lists gracefully

Example

import time

class MyProcessor(ShardRecordProcessor):
    def __init__(self):
        self.last_process_time = time.time()

    async def initialize(self, input: InitializationInput) -> None:
        pass

    async def process_records(self, input: ProcessRecordsInput) -> None:
        # This is called even for empty batches when the setting is enabled
        if not input.records:
            idle_time = time.time() - self.last_process_time
            if idle_time > 60:
                print(f"No records received for {idle_time:.0f}s")
        else:
            for record in input.records:
                # Process the record
                pass
        
        self.last_process_time = time.time()

    async def lease_lost(self, input: LeaseLostInput) -> None:
        pass

    async def shard_ended(self, input: ShardEndedInput) -> None:
        await input.checkpointer.checkpoint()

    async def shutdown_requested(self, input: ShutdownRequestedInput) -> None:
        await input.checkpointer.checkpoint()

Applies to Both Retrieval Modes

This setting works with both Polling (DEFAULT) and Enhanced Fan-Out (EFO) retrieval modes. The behavior is identical regardless of how records are fetched from Kinesis.

Enhanced Fan-Out (EFO)

Enhanced Fan-Out is an optional retrieval mode that provides dedicated throughput for each consumer. This is particularly useful for high-throughput applications or when multiple consumers need to read from the same stream.

Comparison: Polling vs Enhanced Fan-Out

Feature	Polling (DEFAULT)	Enhanced Fan-Out (FANOUT)
Throughput	2 MB/s shared per shard	2 MB/s dedicated per consumer per shard
Latency	~200ms+	~70ms
API	GetRecords	SubscribeToShard (HTTP/2 push)
Consumers per shard	Limited by shared throughput	Up to 20 (or 50 with EFO Advantage)
Cost	Standard Kinesis pricing	Additional per-consumer per-shard-hour fee

Enabling Enhanced Fan-Out

from native_aws_kcl_py import (
    RecordFetcherFactory,
    RecordFetcherFactoryConfig,
    RetrievalMode,
    InitialPositionInStreamExtended,
    InitialPositionInStream,
)

# Parse retrieval mode from environment or config
retrieval_mode = RecordFetcherFactory.parse_retrieval_mode("FANOUT")
# Supports: POLLING, FANOUT, FAN_OUT, EFO, ENHANCED_FAN_OUT, DEFAULT

# Create factory config for EFO
config = RecordFetcherFactoryConfig(
    retrieval_mode=RetrievalMode.FANOUT,
    stream_name="my-stream",
    max_records=10000,
    initial_position=InitialPositionInStreamExtended(
        initial_position_in_stream=InitialPositionInStream.TRIM_HORIZON
    ),
    stream_arn="arn:aws:kinesis:eu-west-1:123456789012:stream/my-stream",
    consumer_arn="arn:aws:kinesis:eu-west-1:123456789012:stream/my-stream/consumer/my-consumer:1234567890",
)

# Create the factory with a Kinesis client
factory = RecordFetcherFactory(config, kinesis_client=kinesis_client)

# Create a fetcher for a specific shard
fetcher = factory.create_fetcher("shardId-000000000001")

# Initialize and fetch records
await fetcher.initialize()
result = await fetcher.fetch_records()

Using EFO with KinesisConsumerConfig

from native_aws_kcl_py import (
    KinesisConsumerConfig,
    AwsConfig,
    RetrievalConfig,
    RetrievalMode,
)

config = KinesisConsumerConfig(
    aws=AwsConfig(region="eu-west-1"),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    retrieval=RetrievalConfig(
        retrieval_mode=RetrievalMode.FANOUT,
    ),
)

When to Use Enhanced Fan-Out

Use EFO when:

• You need dedicated throughput per consumer
• You have multiple consumers reading from the same stream
• You need lower latency (~70ms vs ~200ms+)
• Your application is throughput-sensitive

Don't use EFO when:

• You have a single consumer per stream (polling is sufficient)
• Cost is a primary concern (EFO has additional per-consumer fees)
• You're using LocalStack for local development (EFO is not supported in LocalStack)

EFO Metrics

When using EFO, additional metrics are available:

Metric	Level	Description
SubscribeToShard.Success	DETAILED	Number of successful SubscribeToShard operations
SubscribeToShard.Time	DETAILED	Time taken for SubscribeToShard operation
SubscribeToShard.Failure	DETAILED	Number of failed SubscribeToShard operations

Resharding

The consumer automatically handles resharding (shard splits and merges):

• Split: One parent shard becomes two child shards
• Merge: Two parent shards become one child shard

Child shards are only processed after all parent shards complete:

╔══════════════════════════════════════════════════════════════════╗
║                 SHARD END DETECTED - RESHARDING                  ║
╠══════════════════════════════════════════════════════════════════╣
║ Parent Shard: shardId-000000000001                               ║
║ Status: Shard has reached its end (no more records)              ║
║ Resharding Type: SPLIT                                           ║
║ Child Shards:                                                    ║
║   • shardId-000000000002                                         ║
║   • shardId-000000000003                                         ║
╚══════════════════════════════════════════════════════════════════╝

Pluggable Persistence Layer

By default, the library uses DynamoDB for storing lease and checkpoint information. However, the persistence layer is abstracted behind the LeaseRefresher interface, allowing you to use different storage backends.

Supported Storage Backends

Backend	Status	Extra Package
DynamoDB	✅ Default	None (built-in)
S3	✅ Supported	None (built-in with aioboto3)
Redis	✅ Supported	aioredis
PostgreSQL	✅ Supported	asyncpg
MySQL	✅ Supported	aiomysql
MongoDB	✅ Supported	motor
Custom	✅ Supported	User-provided

Using a Custom Persistence Backend

You can implement the LeaseRefresher interface to create your own persistence backend:

from native_aws_kcl_py import (
    LeaseRefresher,
    Lease,
    KinesisConsumerConfig,
    AwsConfig,
    Scheduler,
)

# Implement the LeaseRefresher interface
class MyCustomLeaseRefresher(LeaseRefresher):
    async def initialize(self) -> None:
        # Initialize your storage backend
        pass
    
    async def create_lease_table_if_not_exists(self) -> bool:
        # Create storage if needed
        pass
    
    async def list_leases(self) -> list[Lease]:
        # List all leases
        pass
    
    async def create_lease_if_not_exists(self, lease: Lease) -> bool:
        # Create a new lease
        pass
    
    async def get_lease(self, lease_key: str) -> Lease | None:
        # Get a specific lease
        pass
    
    async def renew_lease(self, lease: Lease) -> bool:
        # Renew a lease
        pass
    
    async def take_lease(self, lease: Lease, owner: str) -> bool:
        # Take ownership of a lease
        pass
    
    async def evict_lease(self, lease: Lease) -> bool:
        # Release a lease
        pass
    
    async def delete_lease(self, lease: Lease) -> bool:
        # Delete a lease
        pass
    
    async def update_lease(self, lease: Lease) -> bool:
        # Update a lease
        pass

# Use in configuration
config = KinesisConsumerConfig(
    aws=AwsConfig(region="eu-west-1"),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    persistence=MyCustomLeaseRefresher(),
)

scheduler = Scheduler(config)
await scheduler.run()

Optimistic Concurrency Control

All persistence backends implement optimistic concurrency control:

Backend	Mechanism
DynamoDB	Conditional writes with version attributes
S3	ETags with If-Match / If-None-Match headers
Redis	WATCH/MULTI/EXEC transactions
PostgreSQL/MySQL	Version column with atomic updates
MongoDB	findOneAndUpdate with version field

Metrics

The library supports publishing application metrics to AWS CloudWatch (or a custom monitoring backend). By default, metrics are disabled (level: NONE).

Metrics Levels

Level	Description
NONE	No metrics are reported (default)
SUMMARY	Aggregated metrics: RecordsProcessed, MillisBehindLatest, LeasesHeld
DETAILED	All SUMMARY metrics plus per-shard and per-operation metrics

Enabling Metrics

from native_aws_kcl_py import (
    KinesisConsumerConfig,
    AwsConfig,
    MetricsConfig,
)

config = KinesisConsumerConfig(
    aws=AwsConfig(region="eu-west-1"),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    metrics=MetricsConfig(
        level="DETAILED",
        namespace="MyApp/KCL",
    ),
)

Custom Metrics Backend

You can implement your own metrics backend (e.g., Prometheus, Datadog) by implementing the IMetricsFactory interface:

from native_aws_kcl_py import (
    IMetricsFactory,
    IMetricsScope,
    MetricsLevel,
)

class MyPrometheusMetricsFactory(IMetricsFactory):
    def create_scope(self, operation: str) -> IMetricsScope:
        return MyPrometheusScope(operation)
    
    def get_metrics_level(self) -> MetricsLevel:
        return MetricsLevel.DETAILED
    
    def is_enabled(self) -> bool:
        return True
    
    def is_detailed_metrics_enabled(self) -> bool:
        return True
    
    async def shutdown(self) -> None:
        # Flush metrics
        pass

# Use the custom factory
config = KinesisConsumerConfig(
    aws=AwsConfig(region="eu-west-1"),
    application_name="my-consumer-app",
    stream_name="my-kinesis-stream",
    record_processor_factory=MyProcessorFactory(),
    metrics=MetricsConfig(
        custom_factory=MyPrometheusMetricsFactory(),
    ),
)

Error Handling

Exceptions

from native_aws_kcl_py import (
    KinesisClientLibraryException,
    LeaseLostException,
    ShutdownException,
    ProcessingTimeoutException,
    is_retryable_exception,
)

async def process_records(self, input: ProcessRecordsInput) -> None:
    try:
        # Process records...
        await input.checkpointer.checkpoint()
    except LeaseLostException:
        # Lease was lost, stop processing
        return
    except ShutdownException:
        # Consumer is shutting down
        return

Retry with Backoff

from native_aws_kcl_py import (
    calculate_backoff_with_jitter,
    BackoffConfig,
    JitterType,
    BACKOFF_PRESETS,
)

# Calculate backoff delay
delay_ms = calculate_backoff_with_jitter(
    attempt=3,
    config=BackoffConfig(
        base_delay_ms=100,
        max_delay_ms=10000,
        jitter_type=JitterType.EQUAL,
    ),
)

# Or use presets
delay_ms = calculate_backoff_with_jitter(
    attempt=3,
    config=BACKOFF_PRESETS["THROTTLE"],
)

Local Testing with LocalStack

config = KinesisConsumerConfig(
    aws=AwsConfig(
        region="us-east-1",
        kinesis_endpoint="http://localhost:4566",
        dynamodb_endpoint="http://localhost:4566",
    ),
    application_name="test-app",
    stream_name="test-stream",
    record_processor_factory=MyProcessorFactory(),
)

Architecture

┌──────────────────────────────────────────────────────────────┐
│                         Scheduler                             │
│  (Main entry point - coordinates all components)              │
└───────────────────┬──────────────────────────────────────────┘
                    │
     ┌──────────────┼──────────────┐
     │              │              │
     ▼              ▼              ▼
┌─────────┐  ┌─────────────┐  ┌──────────────┐
│ Lease   │  │   Shard     │  │    Shard     │
│Coordin- │  │  Detector   │  │   Syncer     │
│  ator   │  │             │  │              │
└────┬────┘  └──────┬──────┘  └──────┬───────┘
     │              │                │
     │              ▼                │
     │       ┌─────────────┐         │
     │       │   Kinesis   │         │
     │       │   Stream    │         │
     │       └─────────────┘         │
     │                               │
     ▼                               │
┌─────────────────────────────────────────────────────────────┐
│                    Shard Consumers                           │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │   Shard 1    │  │   Shard 2    │  │   Shard 3    │       │
│  │  Consumer    │  │  Consumer    │  │  Consumer    │       │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘       │
│         │                 │                 │                │
│         ▼                 ▼                 ▼                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │   Record     │  │   Record     │  │   Record     │       │
│  │  Processor   │  │  Processor   │  │  Processor   │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
└─────────────────────────────────────────────────────────────┘

API Reference

Scheduler

The main entry point for the KCL.

class Scheduler:
    def __init__(self, config: KinesisConsumerConfig):
        """Initialize the scheduler with configuration."""
        pass
    
    async def run(self) -> None:
        """Start the scheduler and begin processing shards."""
        pass
    
    async def shutdown(self) -> None:
        """Gracefully shutdown the scheduler."""
        pass
    
    def get_state(self) -> SchedulerState:
        """Get the current state of the scheduler."""
        pass
    
    def get_active_consumer_count(self) -> int:
        """Get the number of active shard consumers."""
        pass

ShardRecordProcessor

Interface for processing records from a shard.

class ShardRecordProcessor(ABC):
    @abstractmethod
    async def initialize(self, input: InitializationInput) -> None:
        """Called when a shard is assigned to the worker."""
        pass
    
    @abstractmethod
    async def process_records(self, input: ProcessRecordsInput) -> None:
        """Called to process a batch of records."""
        pass
    
    @abstractmethod
    async def lease_lost(self, input: LeaseLostInput) -> None:
        """Called when the lease is lost to another worker."""
        pass
    
    @abstractmethod
    async def shard_ended(self, input: ShardEndedInput) -> None:
        """Called when the shard reaches its end (resharding)."""
        pass
    
    @abstractmethod
    async def shutdown_requested(self, input: ShutdownRequestedInput) -> None:
        """Called when the worker is shutting down."""
        pass

RecordProcessorCheckpointer

Interface for checkpointing progress.

class RecordProcessorCheckpointer:
    async def checkpoint(
        self,
        sequence_number: str | None = None,
        sub_sequence_number: int | None = None,
    ) -> None:
        """
        Checkpoint at the specified sequence number.
        If not provided, checkpoints at the last record processed.
        """
        pass

Migration from AWS KCL (Java-based)

If you're migrating from the Java-based KCL:

1. No Java Required: This library is pure Python - no JVM needed
2. Same Concepts: Leases, checkpoints, and record processors work the same way
3. Compatible DynamoDB Table: Can reuse existing lease tables (same schema)
4. Async/Await: All operations are async for better performance

Dependencies

• aioboto3 - Async AWS SDK for Python
• electreon-common-utils - Shared Electreon utilities (optional, for ElectreonLogger)

Optional Dependencies

• aioredis - For Redis persistence
• asyncpg - For PostgreSQL persistence
• aiomysql - For MySQL persistence
• motor - For MongoDB persistence

License

UNLICENSED - Proprietary