Queue vs Stream.

# Example: Order events for same user stay in order
producer.send(
    topic="orders",
    key=user_id,        # Partition key
    value=order_event
)

# All events for user_123 go to same partition
# Processed in order by one consumer

# Basic Queue Implementation with Redis
# WARNING: Not production-ready - for learning only

import redis
import json
from typing import Any, Optional

class SimpleRedisQueue:
    """
    Basic queue using Redis lists.
    Messages are pushed to the right, popped from the left (FIFO).
    """
    
    def __init__(self, redis_url: str, queue_name: str):
        self.redis = redis.from_url(redis_url)
        self.queue_name = queue_name
    
    def publish(self, message: dict) -> None:
        """
        Add a message to the queue.
        """
        serialized = json.dumps(message)
        self.redis.rpush(self.queue_name, serialized)
    
    def consume(self, timeout: int = 0) -> Optional[dict]:
        """
        Get and remove a message from the queue.
        Blocks until a message is available or timeout.
        
        Args:
            timeout: Seconds to wait (0 = wait forever)
            
        Returns:
            Message dict or None if timeout
        """
        # BLPOP blocks until message available
        result = self.redis.blpop(self.queue_name, timeout=timeout)
        
        if result is None:
            return None
        
        _, message = result
        return json.loads(message)


# Usage example
def producer_example():
    queue = SimpleRedisQueue("redis://localhost:6379", "orders")
    
    # Publish some orders
    queue.publish({"order_id": "123", "user": "alice", "amount": 99.99})
    queue.publish({"order_id": "124", "user": "bob", "amount": 49.99})
    
    print("Published 2 orders")


def consumer_example():
    queue = SimpleRedisQueue("redis://localhost:6379", "orders")
    
    while True:
        message = queue.consume(timeout=5)
        
        if message is None:
            print("No messages, waiting...")
            continue
        
        print(f"Processing order: {message['order_id']}")
        # Process the order...
        
        # NOTE: If we crash here, message is LOST!
        # No acknowledgment, no retry.

# Basic Kafka Producer/Consumer
# WARNING: Not production-ready - for learning only

from kafka import KafkaProducer, KafkaConsumer
import json

class SimpleKafkaProducer:
    """
    Basic Kafka producer.
    """
    
    def __init__(self, bootstrap_servers: str):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None
        )
    
    def publish(self, topic: str, message: dict, key: str = None) -> None:
        """
        Send message to topic.
        Key determines which partition receives the message.
        """
        self.producer.send(topic, value=message, key=key)
        self.producer.flush()  # Wait for send to complete


class SimpleKafkaConsumer:
    """
    Basic Kafka consumer.
    """
    
    def __init__(self, bootstrap_servers: str, topic: str, group_id: str):
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='earliest',  # Start from beginning if new
            enable_auto_commit=True,        # Automatically commit offsets
            value_deserializer=lambda v: json.loads(v.decode('utf-8'))
        )
    
    def consume(self):
        """
        Yield messages from the topic.
        """
        for message in self.consumer:
            yield {
                'topic': message.topic,
                'partition': message.partition,
                'offset': message.offset,
                'key': message.key.decode('utf-8') if message.key else None,
                'value': message.value
            }


# Usage example
def kafka_producer_example():
    producer = SimpleKafkaProducer("localhost:9092")
    
    # Messages with same key go to same partition (ordered)
    producer.publish("orders", {"order_id": "123", "action": "created"}, key="user_alice")
    producer.publish("orders", {"order_id": "123", "action": "paid"}, key="user_alice")
    producer.publish("orders", {"order_id": "124", "action": "created"}, key="user_bob")


def kafka_consumer_example():
    consumer = SimpleKafkaConsumer(
        "localhost:9092", 
        "orders", 
        group_id="order-processor"
    )
    
    for message in consumer.consume():
        print(f"Partition {message['partition']}, Offset {message['offset']}")
        print(f"Key: {message['key']}, Value: {message['value']}")
        # Process the message...
        
        # NOTE: With auto_commit=True, offset is committed automatically
        # If we crash, we might reprocess (at-least-once)

# Production-Ready Kafka Consumer
# With proper error handling, retries, and monitoring

import asyncio
from dataclasses import dataclass, field
from typing import Callable, Optional, Dict, Any, List
from datetime import datetime, timedelta, timezone
from enum import Enum
import logging
import json
from contextlib import asynccontextmanager

from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from aiokafka.errors import KafkaError

logger = logging.getLogger(__name__)


class ProcessingResult(Enum):
    SUCCESS = "success"
    RETRY = "retry"           # Transient failure, retry later
    DEAD_LETTER = "dead_letter"  # Permanent failure, send to DLQ


@dataclass
class ConsumerConfig:
    """Configuration for Kafka consumer."""
    bootstrap_servers: str
    topic: str
    group_id: str
    
    # Retry configuration
    max_retries: int = 3
    retry_backoff_ms: int = 1000
    max_retry_backoff_ms: int = 30000
    
    # Consumer configuration
    auto_offset_reset: str = "earliest"
    enable_auto_commit: bool = False  # Manual commit for reliability
    max_poll_records: int = 100
    session_timeout_ms: int = 30000
    
    # Dead letter topic
    dead_letter_topic: Optional[str] = None


@dataclass
class MessageContext:
    """Context for message processing."""
    topic: str
    partition: int
    offset: int
    key: Optional[str]
    value: Dict[str, Any]
    timestamp: datetime
    headers: Dict[str, str] = field(default_factory=dict)
    retry_count: int = 0


class KafkaConsumerService:
    """
    Production-ready Kafka consumer with:
    - Manual offset commits (at-least-once delivery)
    - Retry with exponential backoff
    - Dead letter queue for failed messages
    - Graceful shutdown
    - Metrics and logging
    """
    
    def __init__(
        self,
        config: ConsumerConfig,
        message_handler: Callable[[MessageContext], ProcessingResult]
    ):
        self.config = config
        self.message_handler = message_handler
        self.consumer: Optional[AIOKafkaConsumer] = None
        self.dlq_producer: Optional[AIOKafkaProducer] = None
        self._running = False
        self._shutdown_event = asyncio.Event()
        
        # Metrics
        self.messages_processed = 0
        self.messages_failed = 0
        self.messages_to_dlq = 0
    
    async def start(self) -> None:
        """Start the consumer."""
        logger.info(f"Starting consumer for topic {self.config.topic}")
        
        self.consumer = AIOKafkaConsumer(
            self.config.topic,
            bootstrap_servers=self.config.bootstrap_servers,
            group_id=self.config.group_id,
            auto_offset_reset=self.config.auto_offset_reset,
            enable_auto_commit=self.config.enable_auto_commit,
            max_poll_records=self.config.max_poll_records,
            session_timeout_ms=self.config.session_timeout_ms,
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            key_deserializer=lambda k: k.decode('utf-8') if k else None
        )
        
        await self.consumer.start()
        
        # Start DLQ producer if configured
        if self.config.dead_letter_topic:
            self.dlq_producer = AIOKafkaProducer(
                bootstrap_servers=self.config.bootstrap_servers,
                value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                key_serializer=lambda k: k.encode('utf-8') if k else None
            )
            await self.dlq_producer.start()
        
        self._running = True
        logger.info(f"Consumer started for topic {self.config.topic}")
    
    async def stop(self) -> None:
        """Gracefully stop the consumer."""
        logger.info("Stopping consumer...")
        self._running = False
        self._shutdown_event.set()
        
        if self.consumer:
            await self.consumer.stop()
        
        if self.dlq_producer:
            await self.dlq_producer.stop()
        
        logger.info(f"Consumer stopped. Processed: {self.messages_processed}, "
                   f"Failed: {self.messages_failed}, DLQ: {self.messages_to_dlq}")
    
    async def run(self) -> None:
        """Main consumer loop."""
        try:
            async for message in self.consumer:
                if not self._running:
                    break
                
                ctx = MessageContext(
                    topic=message.topic,
                    partition=message.partition,
                    offset=message.offset,
                    key=message.key,
                    value=message.value,
                    timestamp=datetime.fromtimestamp(message.timestamp / 1000, timezone.utc),
                    headers={k: v.decode('utf-8') for k, v in (message.headers or [])},
                    retry_count=self._get_retry_count(message.headers)
                )
                
                await self._process_message(ctx)
                
        except asyncio.CancelledError:
            logger.info("Consumer loop cancelled")
        except Exception as e:
            logger.error(f"Consumer loop error: {e}", exc_info=True)
            raise
    
    async def _process_message(self, ctx: MessageContext) -> None:
        """Process a single message with retry logic."""
        try:
            result = await self._execute_handler(ctx)
            
            if result == ProcessingResult.SUCCESS:
                # Commit offset on success
                await self.consumer.commit()
                self.messages_processed += 1
                logger.debug(f"Processed message at offset {ctx.offset}")
                
            elif result == ProcessingResult.RETRY:
                await self._handle_retry(ctx)
                
            elif result == ProcessingResult.DEAD_LETTER:
                await self._send_to_dlq(ctx, "Handler returned DEAD_LETTER")
                await self.consumer.commit()
                
        except Exception as e:
            logger.error(f"Error processing message: {e}", exc_info=True)
            await self._handle_retry(ctx)
    
    async def _execute_handler(self, ctx: MessageContext) -> ProcessingResult:
        """Execute the message handler with error handling."""
        try:
            # Handler might be sync or async
            result = self.message_handler(ctx)
            if asyncio.iscoroutine(result):
                result = await result
            return result
        except Exception as e:
            logger.warning(f"Handler raised exception: {e}")
            return ProcessingResult.RETRY
    
    async def _handle_retry(self, ctx: MessageContext) -> None:
        """Handle message retry with exponential backoff."""
        if ctx.retry_count >= self.config.max_retries:
            logger.warning(f"Max retries ({self.config.max_retries}) exceeded "
                          f"for message at offset {ctx.offset}")
            await self._send_to_dlq(ctx, f"Max retries exceeded after {ctx.retry_count} attempts")
            await self.consumer.commit()
            return
        
        # Calculate backoff
        backoff_ms = min(
            self.config.retry_backoff_ms * (2 ** ctx.retry_count),
            self.config.max_retry_backoff_ms
        )
        
        logger.info(f"Retrying message at offset {ctx.offset} "
                   f"(attempt {ctx.retry_count + 1}/{self.config.max_retries}) "
                   f"after {backoff_ms}ms")
        
        await asyncio.sleep(backoff_ms / 1000)
        
        # Re-process with incremented retry count
        ctx.retry_count += 1
        await self._process_message(ctx)
    
    async def _send_to_dlq(self, ctx: MessageContext, reason: str) -> None:
        """Send failed message to dead letter queue."""
        if not self.dlq_producer or not self.config.dead_letter_topic:
            logger.warning(f"No DLQ configured, dropping message at offset {ctx.offset}")
            self.messages_failed += 1
            return
        
        dlq_message = {
            "original_topic": ctx.topic,
            "original_partition": ctx.partition,
            "original_offset": ctx.offset,
            "original_key": ctx.key,
            "original_value": ctx.value,
            "original_timestamp": ctx.timestamp.isoformat(),
            "failure_reason": reason,
            "retry_count": ctx.retry_count,
            "failed_at": datetime.now(timezone.utc).isoformat()
        }
        
        await self.dlq_producer.send_and_wait(
            self.config.dead_letter_topic,
            value=dlq_message,
            key=ctx.key
        )
        
        self.messages_to_dlq += 1
        logger.info(f"Sent message at offset {ctx.offset} to DLQ: {reason}")
    
    def _get_retry_count(self, headers: Optional[List]) -> int:
        """Extract retry count from message headers."""
        if not headers:
            return 0
        for key, value in headers:
            if key == "retry_count":
                return int(value.decode('utf-8'))
        return 0


# =============================================================================
# Example Usage
# =============================================================================

async def order_handler(ctx: MessageContext) -> ProcessingResult:
    """Example message handler for order events."""
    try:
        order = ctx.value
        logger.info(f"Processing order {order.get('order_id')} "
                   f"for user {ctx.key}")
        
        # Simulate processing
        if order.get("amount", 0) < 0:
            # Invalid order - send to DLQ
            logger.warning(f"Invalid order amount: {order.get('amount')}")
            return ProcessingResult.DEAD_LETTER
        
        if order.get("action") == "fail_temporary":
            # Simulate transient failure
            return ProcessingResult.RETRY
        
        # Process the order...
        await asyncio.sleep(0.1)  # Simulate work
        
        return ProcessingResult.SUCCESS
        
    except KeyError as e:
        # Missing required field - permanent failure
        logger.error(f"Missing required field: {e}")
        return ProcessingResult.DEAD_LETTER
    except Exception as e:
        # Unknown error - retry
        logger.error(f"Processing error: {e}")
        return ProcessingResult.RETRY


async def main():
    """Example main function."""
    config = ConsumerConfig(
        bootstrap_servers="localhost:9092",
        topic="orders",
        group_id="order-processor",
        dead_letter_topic="orders-dlq",
        max_retries=3,
        retry_backoff_ms=1000
    )
    
    consumer = KafkaConsumerService(config, order_handler)
    
    try:
        await consumer.start()
        await consumer.run()
    except KeyboardInterrupt:
        pass
    finally:
        await consumer.stop()


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

# Production-Ready Kafka Producer
# With batching, retries, and delivery confirmation

import asyncio
from dataclasses import dataclass
from typing import Optional, Dict, Any, Callable
import logging
import json
from datetime import datetime, timezone

from aiokafka import AIOKafkaProducer
from aiokafka.errors import KafkaError

logger = logging.getLogger(__name__)


@dataclass
class ProducerConfig:
    """Configuration for Kafka producer."""
    bootstrap_servers: str
    
    # Reliability settings
    acks: str = "all"                    # Wait for all replicas
    request_timeout_ms: int = 30000      # aiokafka retries a send until this timeout
    retry_backoff_ms: int = 100          # Backoff between retries
    
    # Batching settings (for throughput)
    linger_ms: int = 5                   # Wait up to 5ms to batch
    batch_size: int = 16384              # 16KB batch size
    
    # Idempotence (prevent duplicates on retry)
    enable_idempotence: bool = True


class KafkaProducerService:
    """
    Production-ready Kafka producer with:
    - Idempotent delivery (no duplicates)
    - Delivery confirmation
    - Batching for throughput
    - Proper error handling
    """
    
    def __init__(self, config: ProducerConfig):
        self.config = config
        self.producer: Optional[AIOKafkaProducer] = None
        
        # Metrics
        self.messages_sent = 0
        self.messages_failed = 0
    
    async def start(self) -> None:
        """Start the producer."""
        self.producer = AIOKafkaProducer(
            bootstrap_servers=self.config.bootstrap_servers,
            acks=self.config.acks,
            request_timeout_ms=self.config.request_timeout_ms,
            retry_backoff_ms=self.config.retry_backoff_ms,
            linger_ms=self.config.linger_ms,
            max_batch_size=self.config.batch_size,
            enable_idempotence=self.config.enable_idempotence,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None
        )
        await self.producer.start()
        logger.info("Producer started")
    
    async def stop(self) -> None:
        """Stop the producer, flushing pending messages."""
        if self.producer:
            await self.producer.stop()
        logger.info(f"Producer stopped. Sent: {self.messages_sent}, "
                   f"Failed: {self.messages_failed}")
    
    async def send(
        self,
        topic: str,
        value: Dict[str, Any],
        key: Optional[str] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> bool:
        """
        Send a message to Kafka.
        
        Args:
            topic: Target topic
            value: Message payload
            key: Partition key (messages with same key go to same partition)
            headers: Optional message headers
            
        Returns:
            True if sent successfully, False otherwise
        """
        try:
            # Convert headers to Kafka format
            kafka_headers = None
            if headers:
                kafka_headers = [(k, v.encode('utf-8')) for k, v in headers.items()]
            
            # Send and wait for acknowledgment
            result = await self.producer.send_and_wait(
                topic,
                value=value,
                key=key,
                headers=kafka_headers
            )
            
            self.messages_sent += 1
            logger.debug(f"Sent message to {topic}:{result.partition}@{result.offset}")
            return True
            
        except KafkaError as e:
            self.messages_failed += 1
            logger.error(f"Failed to send message: {e}")
            return False
    
    async def send_batch(
        self,
        topic: str,
        messages: list[tuple[Optional[str], Dict[str, Any]]]
    ) -> int:
        """
        Send multiple messages efficiently.
        
        Args:
            topic: Target topic
            messages: List of (key, value) tuples
            
        Returns:
            Number of successfully sent messages
        """
        tasks = []
        for key, value in messages:
            # Don't wait for each message individually
            future = await self.producer.send(topic, value=value, key=key)
            tasks.append(future)
        
        # Wait for all to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        success_count = sum(1 for r in results if not isinstance(r, Exception))
        self.messages_sent += success_count
        self.messages_failed += len(results) - success_count
        
        return success_count


# =============================================================================
# Example Usage
# =============================================================================

async def producer_example():
    """Example of using the producer."""
    config = ProducerConfig(
        bootstrap_servers="localhost:9092",
        acks="all",
        enable_idempotence=True
    )
    
    producer = KafkaProducerService(config)
    await producer.start()
    
    try:
        # Send single message
        await producer.send(
            topic="orders",
            key="user_123",
            value={
                "order_id": "order_456",
                "action": "created",
                "amount": 99.99,
                "timestamp": datetime.now(timezone.utc).isoformat()
            },
            headers={"source": "api", "version": "1"}
        )
        
        # Send batch
        messages = [
            ("user_123", {"order_id": "order_457", "action": "created"}),
            ("user_456", {"order_id": "order_458", "action": "created"}),
            ("user_789", {"order_id": "order_459", "action": "created"}),
        ]
        sent = await producer.send_batch("orders", messages)
        print(f"Sent {sent} messages")
        
    finally:
        await producer.stop()

class IdempotentOrderProcessor:
    """
    Process orders idempotently using a processed_ids table.
    """
    
    def __init__(self, db_pool):
        self.db = db_pool
    
    async def process_order(self, ctx: MessageContext) -> ProcessingResult:
        """Process order with idempotency check."""
        order = ctx.value
        message_id = f"{ctx.topic}:{ctx.partition}:{ctx.offset}"
        
        async with self.db.acquire() as conn:
            async with conn.transaction():
                # Check if already processed
                existing = await conn.fetchval(
                    "SELECT 1 FROM processed_messages WHERE message_id = $1",
                    message_id
                )
                
                if existing:
                    logger.info(f"Message {message_id} already processed, skipping")
                    return ProcessingResult.SUCCESS
                
                # Process the order
                await conn.execute(
                    """
                    INSERT INTO orders (order_id, user_id, amount, status)
                    VALUES ($1, $2, $3, 'created')
                    """,
                    order['order_id'], order['user_id'], order['amount']
                )
                
                # Mark as processed (in same transaction!)
                await conn.execute(
                    """
                    INSERT INTO processed_messages (message_id, processed_at)
                    VALUES ($1, NOW())
                    """,
                    message_id
                )
        
        return ProcessingResult.SUCCESS

from aiokafka import ConsumerRebalanceListener

class GracefulRebalanceListener(ConsumerRebalanceListener):
    """
    Handle partition rebalance gracefully.
    Commit offsets before partitions are revoked.
    """
    
    def __init__(self, consumer: AIOKafkaConsumer, processor):
        self.consumer = consumer
        self.processor = processor
    
    async def on_partitions_revoked(self, revoked):
        """Called before partitions are taken away."""
        logger.info(f"Partitions being revoked: {revoked}")
        
        # Finish any in-progress work
        await self.processor.finish_current_batch()
        
        # Commit current offsets
        await self.consumer.commit()
        
        logger.info("Committed offsets before rebalance")
    
    async def on_partitions_assigned(self, assigned):
        """Called after new partitions are assigned."""
        logger.info(f"Partitions assigned: {assigned}")
        
        # Optionally seek to specific offsets
        # for tp in assigned:
        #     await self.consumer.seek(tp, specific_offset)

from dataclasses import dataclass
from typing import Dict
import asyncio

@dataclass
class LagMetrics:
    partition: int
    current_offset: int
    end_offset: int
    lag: int


class ConsumerLagMonitor:
    """Monitor consumer lag and alert when thresholds exceeded."""
    
    def __init__(
        self,
        consumer: AIOKafkaConsumer,
        warning_threshold: int = 10000,
        critical_threshold: int = 100000
    ):
        self.consumer = consumer
        self.warning_threshold = warning_threshold
        self.critical_threshold = critical_threshold
    
    async def get_lag(self) -> Dict[int, LagMetrics]:
        """Get lag for all assigned partitions."""
        lag_metrics = {}
        
        # Get end offsets (latest)
        partitions = self.consumer.assignment()
        end_offsets = await self.consumer.end_offsets(partitions)
        
        for tp in partitions:
            current = await self.consumer.position(tp)
            end = end_offsets[tp]
            lag = end - current
            
            lag_metrics[tp.partition] = LagMetrics(
                partition=tp.partition,
                current_offset=current,
                end_offset=end,
                lag=lag
            )
        
        return lag_metrics
    
    async def check_lag(self) -> None:
        """Check lag and log warnings/errors."""
        lag_metrics = await self.get_lag()
        
        total_lag = sum(m.lag for m in lag_metrics.values())
        
        if total_lag >= self.critical_threshold:
            logger.error(f"CRITICAL: Consumer lag is {total_lag:,} messages!")
            # Send alert to PagerDuty/Slack
        elif total_lag >= self.warning_threshold:
            logger.warning(f"WARNING: Consumer lag is {total_lag:,} messages")
        else:
            logger.debug(f"Consumer lag: {total_lag:,} messages")
        
        # Log per-partition metrics
        for partition, metrics in lag_metrics.items():
            logger.debug(f"Partition {partition}: lag={metrics.lag:,}")
    
    async def monitor_loop(self, interval_seconds: int = 30):
        """Continuously monitor lag."""
        while True:
            try:
                await self.check_lag()
            except Exception as e:
                logger.error(f"Error checking lag: {e}")
            
            await asyncio.sleep(interval_seconds)