Metadata-Version: 2.4
Name: electreonwireless-native-aws-kcl-py
Version: 0.1.2
Summary: Native AWS Kinesis Consumer Library for Python - KCL implementation with lease coordination, resharding, and metrics
License: MIT
License-File: LICENSE
Keywords: aws,kinesis,consumer,kcl,streaming,async
Author: Electreon Cloud Team
Requires-Python: >=3.10,<4.0
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Provides-Extra: all
Provides-Extra: mongodb
Provides-Extra: mysql
Provides-Extra: postgresql
Provides-Extra: redis
Requires-Dist: aioboto3 (>=13.0,<14.0)
Requires-Dist: aiobotocore (>=2.11.0,<3.0.0)
Requires-Dist: aiohttp (>=3.8.5,<4.0.0)
Requires-Dist: aiomysql (>=0.2,<0.3) ; extra == "mysql" or extra == "all"
Requires-Dist: asyncpg (>=0.29,<0.30) ; extra == "postgresql" or extra == "all"
Requires-Dist: motor (>=3.3,<4.0) ; extra == "mongodb" or extra == "all"
Requires-Dist: redis (>=5.0,<6.0) ; extra == "redis" or extra == "all"
Project-URL: Repository, https://github.com/electreonwireless/native-aws-kcl-py
Description-Content-Type: text/markdown

# 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)

```bash
pip install native-aws-kcl-py
```

Or with Poetry:

```bash
poetry add native-aws-kcl-py
```

### With Optional Persistence Backends

Install with additional drivers for your chosen persistence backend:

```bash
# 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

```python
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

```python
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

```python
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

```python
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

```python
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:

```bash
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

```python
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

```python
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

```python
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:

```python
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

```python
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:

```python
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

```python
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

```python
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

```python
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.

```python
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.

```python
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.

```python
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

