Backpressure and Flow Control.

# Backpressure Detection System

import asyncio
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional, List
from enum import Enum
import logging

logger = logging.getLogger(__name__)


class BackpressureLevel(Enum):
    NONE = "none"           # All clear
    WARNING = "warning"     # Early signs, monitor closely
    CRITICAL = "critical"   # Take action now
    EMERGENCY = "emergency" # System at risk of failure


@dataclass
class BackpressureConfig:
    """Thresholds for backpressure detection."""
    # Lag thresholds
    lag_warning: int = 10_000
    lag_critical: int = 100_000
    lag_emergency: int = 1_000_000
    
    # Queue depth thresholds  
    depth_warning: int = 50_000
    depth_critical: int = 200_000
    depth_emergency: int = 1_000_000
    
    # Latency thresholds (seconds)
    latency_warning: float = 5.0
    latency_critical: float = 30.0
    latency_emergency: float = 300.0
    
    # Rate of change thresholds
    lag_growth_rate_warning: float = 1000.0  # messages/sec growth


@dataclass
class BackpressureStatus:
    """Current backpressure status."""
    level: BackpressureLevel
    consumer_lag: int
    queue_depth: int
    processing_latency_p99: float
    lag_growth_rate: float
    timestamp: datetime
    details: str


class BackpressureMonitor:
    """
    Monitors system metrics and detects backpressure conditions.
    """
    
    def __init__(
        self,
        config: BackpressureConfig,
        metrics_client,  # Your metrics system (Prometheus, Datadog, etc.)
    ):
        self.config = config
        self.metrics = metrics_client
        self._lag_history: List[tuple] = []  # (timestamp, lag)
    
    async def check(self) -> BackpressureStatus:
        """Check current backpressure status."""
        # Gather metrics
        lag = await self.metrics.get_consumer_lag()
        depth = await self.metrics.get_queue_depth()
        latency = await self.metrics.get_processing_latency_p99()
        
        # Calculate lag growth rate
        growth_rate = self._calculate_lag_growth_rate(lag)
        
        # Determine level
        level = self._determine_level(lag, depth, latency, growth_rate)
        
        # Build status
        status = BackpressureStatus(
            level=level,
            consumer_lag=lag,
            queue_depth=depth,
            processing_latency_p99=latency,
            lag_growth_rate=growth_rate,
            timestamp=datetime.now(timezone.utc),
            details=self._build_details(lag, depth, latency, growth_rate)
        )
        
        # Log if not healthy
        if level != BackpressureLevel.NONE:
            logger.warning(f"Backpressure detected: {status}")
        
        return status
    
    def _determine_level(
        self, lag: int, depth: int, latency: float, growth_rate: float
    ) -> BackpressureLevel:
        """Determine backpressure level from metrics."""
        
        # Emergency: Any metric at emergency level
        if (lag >= self.config.lag_emergency or
            depth >= self.config.depth_emergency or
            latency >= self.config.latency_emergency):
            return BackpressureLevel.EMERGENCY
        
        # Critical: Any metric at critical level
        if (lag >= self.config.lag_critical or
            depth >= self.config.depth_critical or
            latency >= self.config.latency_critical):
            return BackpressureLevel.CRITICAL
        
        # Warning: Any metric at warning level OR lag growing fast
        if (lag >= self.config.lag_warning or
            depth >= self.config.depth_warning or
            latency >= self.config.latency_warning or
            growth_rate >= self.config.lag_growth_rate_warning):
            return BackpressureLevel.WARNING
        
        return BackpressureLevel.NONE
    
    def _calculate_lag_growth_rate(self, current_lag: int) -> float:
        """Calculate how fast lag is growing (messages/sec)."""
        now = datetime.now(timezone.utc)
        self._lag_history.append((now, current_lag))
        
        # Keep only last 60 seconds
        cutoff = now - timedelta(seconds=60)
        self._lag_history = [
            (t, l) for t, l in self._lag_history if t > cutoff
        ]
        
        if len(self._lag_history) < 2:
            return 0.0
        
        # Calculate rate from oldest to newest
        oldest_time, oldest_lag = self._lag_history[0]
        newest_time, newest_lag = self._lag_history[-1]
        
        time_diff = (newest_time - oldest_time).total_seconds()
        if time_diff == 0:
            return 0.0
        
        lag_diff = newest_lag - oldest_lag
        return lag_diff / time_diff
    
    def _build_details(
        self, lag: int, depth: int, latency: float, growth_rate: float
    ) -> str:
        """Build human-readable details string."""
        parts = []
        
        if lag >= self.config.lag_warning:
            parts.append(f"lag={lag:,}")
        if depth >= self.config.depth_warning:
            parts.append(f"depth={depth:,}")
        if latency >= self.config.latency_warning:
            parts.append(f"latency_p99={latency:.1f}s")
        if growth_rate >= self.config.lag_growth_rate_warning:
            parts.append(f"lag_growth={growth_rate:.0f}/sec")
        
        return ", ".join(parts) if parts else "all metrics healthy"

# Load Shedding Implementation

from enum import IntEnum
import random


class Priority(IntEnum):
    CRITICAL = 0   # Never drop (payments, orders)
    HIGH = 1       # Drop only in emergency
    MEDIUM = 2     # Drop in critical+
    LOW = 3        # Drop in warning+


class LoadShedder:
    """
    Drops messages based on priority and backpressure level.
    """
    
    # Which priorities to shed at each level
    SHED_MATRIX = {
        BackpressureLevel.NONE: set(),
        BackpressureLevel.WARNING: {Priority.LOW},
        BackpressureLevel.CRITICAL: {Priority.LOW, Priority.MEDIUM},
        BackpressureLevel.EMERGENCY: {Priority.LOW, Priority.MEDIUM, Priority.HIGH},
        # CRITICAL priority is NEVER shed
    }
    
    def __init__(self, monitor: BackpressureMonitor):
        self.monitor = monitor
        self.shed_count = 0
        self.total_count = 0
    
    async def should_process(self, message: dict) -> bool:
        """
        Decide whether to process or shed this message.
        
        Returns True if message should be processed.
        """
        self.total_count += 1
        
        # Get current backpressure level
        status = await self.monitor.check()
        
        # Get message priority
        priority = Priority(message.get('priority', Priority.MEDIUM))
        
        # Check if this priority should be shed
        priorities_to_shed = self.SHED_MATRIX[status.level]
        
        if priority in priorities_to_shed:
            self.shed_count += 1
            logger.debug(
                f"Shedding message (priority={priority.name}, "
                f"level={status.level.name})"
            )
            return False
        
        return True
    
    def get_shed_rate(self) -> float:
        """Get percentage of messages being shed."""
        if self.total_count == 0:
            return 0.0
        return self.shed_count / self.total_count * 100

# Adaptive Rate Limiter

import asyncio
import time
from dataclasses import dataclass


@dataclass
class RateLimitConfig:
    """Configuration for adaptive rate limiting."""
    base_rate: float = 1000.0          # Normal rate (messages/sec)
    min_rate: float = 100.0            # Minimum rate under pressure
    max_rate: float = 5000.0           # Maximum rate when healthy
    
    # Lag thresholds for adjustment
    lag_low: int = 1000                # Below this: increase rate
    lag_high: int = 10000              # Above this: decrease rate
    
    adjustment_factor: float = 0.1     # How much to adjust (10%)


class AdaptiveRateLimiter:
    """
    Rate limiter that adjusts based on consumer lag.
    
    When lag is low: allow higher rate
    When lag is high: reduce rate to let consumer catch up
    """
    
    def __init__(
        self,
        config: RateLimitConfig,
        monitor: BackpressureMonitor
    ):
        self.config = config
        self.monitor = monitor
        self.current_rate = config.base_rate
        self.tokens = config.base_rate
        self.last_update = time.monotonic()
        self._lock = asyncio.Lock()
    
    async def acquire(self, timeout: float = 10.0) -> bool:
        """
        Acquire permission to send a message.
        
        Returns True if acquired, False if timed out.
        """
        deadline = time.monotonic() + timeout
        
        while time.monotonic() < deadline:
            async with self._lock:
                self._refill_tokens()
                
                if self.tokens >= 1:
                    self.tokens -= 1
                    return True
            
            # Wait a bit before retrying
            await asyncio.sleep(0.01)
        
        return False
    
    async def adjust_rate(self):
        """Adjust rate based on current lag. Call periodically."""
        status = await self.monitor.check()
        lag = status.consumer_lag
        
        async with self._lock:
            if lag < self.config.lag_low:
                # Lag is low, we can go faster
                self.current_rate = min(
                    self.current_rate * (1 + self.config.adjustment_factor),
                    self.config.max_rate
                )
                logger.debug(f"Lag low ({lag}), increasing rate to {self.current_rate:.0f}/sec")
                
            elif lag > self.config.lag_high:
                # Lag is high, slow down
                self.current_rate = max(
                    self.current_rate * (1 - self.config.adjustment_factor),
                    self.config.min_rate
                )
                logger.warning(f"Lag high ({lag}), decreasing rate to {self.current_rate:.0f}/sec")
    
    def _refill_tokens(self):
        """Refill tokens based on elapsed time."""
        now = time.monotonic()
        elapsed = now - self.last_update
        self.last_update = now
        
        # Add tokens based on current rate
        self.tokens = min(
            self.tokens + elapsed * self.current_rate,
            self.current_rate  # Cap at 1 second worth
        )

# Auto-scaling Consumer Group

from kubernetes import client, config as k8s_config


class ConsumerAutoScaler:
    """
    Auto-scales consumer deployment based on lag.
    Uses Kubernetes HPA custom metrics.
    """
    
    def __init__(
        self,
        namespace: str,
        deployment_name: str,
        min_replicas: int = 1,
        max_replicas: int = 10,
        lag_per_replica: int = 10000  # Target lag per consumer
    ):
        k8s_config.load_incluster_config()
        self.apps_v1 = client.AppsV1Api()
        
        self.namespace = namespace
        self.deployment_name = deployment_name
        self.min_replicas = min_replicas
        self.max_replicas = max_replicas
        self.lag_per_replica = lag_per_replica
    
    async def scale_if_needed(self, current_lag: int, partition_count: int):
        """
        Calculate and apply optimal replica count.
        """
        # Get current replicas
        deployment = self.apps_v1.read_namespaced_deployment(
            self.deployment_name, self.namespace
        )
        current_replicas = deployment.spec.replicas
        
        # Calculate desired replicas based on lag
        desired_by_lag = max(1, current_lag // self.lag_per_replica)
        
        # Cap at partition count (can't have more consumers than partitions)
        desired_replicas = min(
            max(self.min_replicas, desired_by_lag),
            min(self.max_replicas, partition_count)
        )
        
        if desired_replicas != current_replicas:
            logger.info(
                f"Scaling {self.deployment_name}: "
                f"{current_replicas} → {desired_replicas} "
                f"(lag={current_lag:,})"
            )
            
            # Apply scale
            deployment.spec.replicas = desired_replicas
            self.apps_v1.patch_namespaced_deployment(
                self.deployment_name,
                self.namespace,
                deployment
            )

# Basic Backpressure-Aware Consumer
# WARNING: Not production-ready - for learning only

import asyncio
from typing import Callable, Any


class SimpleBackpressureConsumer:
    """
    Consumer that monitors its own processing rate
    and signals when falling behind.
    """
    
    def __init__(
        self,
        process_fn: Callable[[dict], Any],
        target_rate: float = 100.0,  # Target messages/sec
        max_queue_size: int = 1000
    ):
        self.process_fn = process_fn
        self.target_rate = target_rate
        self.max_queue_size = max_queue_size
        
        self._queue = asyncio.Queue(maxsize=max_queue_size)
        self._processed_count = 0
        self._dropped_count = 0
    
    async def submit(self, message: dict) -> bool:
        """
        Submit a message for processing.
        Returns False if queue is full (backpressure signal).
        """
        try:
            self._queue.put_nowait(message)
            return True
        except asyncio.QueueFull:
            self._dropped_count += 1
            return False  # Backpressure signal!
    
    async def run(self):
        """Process messages from queue."""
        while True:
            message = await self._queue.get()
            
            try:
                await self.process_fn(message)
                self._processed_count += 1
            except Exception as e:
                print(f"Processing error: {e}")
            finally:
                self._queue.task_done()
    
    @property
    def queue_depth(self) -> int:
        return self._queue.qsize()
    
    @property
    def is_under_pressure(self) -> bool:
        return self._queue.qsize() > self.max_queue_size * 0.8


# Usage
async def process_message(msg):
    await asyncio.sleep(0.01)  # Simulate work
    print(f"Processed: {msg}")

async def main():
    consumer = SimpleBackpressureConsumer(process_message)
    
    # Start consumer
    consumer_task = asyncio.create_task(consumer.run())
    
    # Submit messages, handle backpressure
    for i in range(10000):
        success = await consumer.submit({"id": i})
        
        if not success:
            print(f"Backpressure! Queue full at message {i}")
            await asyncio.sleep(0.1)  # Back off

# Production-Ready Backpressure Management System

import asyncio
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, List, Callable, Set
from enum import Enum
import logging
import time

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class BackpressureManagerConfig:
    """Configuration for backpressure management."""
    
    # Detection thresholds
    lag_warning: int = 10_000
    lag_critical: int = 100_000
    lag_emergency: int = 500_000
    
    queue_depth_warning: int = 50_000
    queue_depth_critical: int = 200_000
    
    # Response configuration
    enable_load_shedding: bool = True
    enable_rate_limiting: bool = True
    enable_auto_scaling: bool = True
    
    # Rate limiting
    base_rate_limit: float = 10_000.0  # messages/sec
    min_rate_limit: float = 1_000.0
    
    # Scaling
    min_consumers: int = 1
    max_consumers: int = 20
    scale_up_threshold: int = 50_000    # Lag to trigger scale up
    scale_down_threshold: int = 5_000   # Lag to trigger scale down
    scale_cooldown_seconds: int = 300   # Wait between scaling actions
    
    # Priorities that can be shed at each level
    shed_on_warning: Set[int] = field(default_factory=lambda: {3})      # LOW only
    shed_on_critical: Set[int] = field(default_factory=lambda: {2, 3})  # MEDIUM, LOW
    shed_on_emergency: Set[int] = field(default_factory=lambda: {1, 2, 3})  # All except CRITICAL


class Priority(Enum):
    CRITICAL = 0   # Never shed (payments, orders)
    HIGH = 1       # Shed only in emergency
    MEDIUM = 2     # Shed in critical+
    LOW = 3        # Shed in warning+


# =============================================================================
# Core Backpressure Manager
# =============================================================================

class BackpressureManager:
    """
    Comprehensive backpressure management system.
    
    Handles:
    - Detection: Monitors lag, queue depth, latency
    - Response: Rate limiting, load shedding, scaling
    - Recovery: Auto-adjusts as pressure subsides
    """
    
    def __init__(
        self,
        config: BackpressureManagerConfig,
        metrics_client,        # Your metrics system
        scaler = None,         # Optional: Kubernetes scaler
    ):
        self.config = config
        self.metrics = metrics_client
        self.scaler = scaler
        
        # State
        self.current_level = BackpressureLevel.NONE
        self.current_rate_limit = config.base_rate_limit
        self.last_scale_time: Optional[datetime] = None
        
        # Rate limiter state
        self._tokens = config.base_rate_limit
        self._last_token_update = time.monotonic()
        self._token_lock = asyncio.Lock()
        
        # Metrics
        self.messages_accepted = 0
        self.messages_shed = 0
        self.messages_rate_limited = 0
    
    async def start(self):
        """Start background monitoring loop."""
        asyncio.create_task(self._monitor_loop())
        asyncio.create_task(self._adjust_loop())
        logger.info("Backpressure manager started")
    
    async def should_accept(self, message: Dict[str, Any]) -> bool:
        """
        Decide whether to accept or reject a message.
        
        Applies rate limiting and load shedding based on current pressure.
        
        Args:
            message: Message dict with optional 'priority' field
            
        Returns:
            True if message should be processed, False if shed/limited
        """
        priority = Priority(message.get('priority', Priority.MEDIUM.value))
        
        # 1. Check load shedding
        if self.config.enable_load_shedding:
            if self._should_shed(priority):
                self.messages_shed += 1
                logger.debug(f"Shedding message (priority={priority.name})")
                return False
        
        # 2. Check rate limiting
        if self.config.enable_rate_limiting:
            acquired = await self._acquire_token()
            if not acquired:
                self.messages_rate_limited += 1
                logger.debug("Rate limited message")
                return False
        
        self.messages_accepted += 1
        return True
    
    def _should_shed(self, priority: Priority) -> bool:
        """Check if this priority should be shed at current level."""
        if self.current_level == BackpressureLevel.NONE:
            return False
        
        if self.current_level == BackpressureLevel.WARNING:
            return priority.value in self.config.shed_on_warning
        
        if self.current_level == BackpressureLevel.CRITICAL:
            return priority.value in self.config.shed_on_critical
        
        if self.current_level == BackpressureLevel.EMERGENCY:
            return priority.value in self.config.shed_on_emergency
        
        return False
    
    async def _acquire_token(self, timeout: float = 0.1) -> bool:
        """Try to acquire a rate limit token."""
        async with self._token_lock:
            self._refill_tokens()
            
            if self._tokens >= 1:
                self._tokens -= 1
                return True
        
        return False
    
    def _refill_tokens(self):
        """Refill tokens based on elapsed time and current rate."""
        now = time.monotonic()
        elapsed = now - self._last_token_update
        self._last_token_update = now
        
        self._tokens = min(
            self._tokens + elapsed * self.current_rate_limit,
            self.current_rate_limit  # Cap at 1 second worth
        )
    
    async def _monitor_loop(self):
        """Background loop to monitor backpressure metrics."""
        while True:
            try:
                await self._update_level()
            except Exception as e:
                logger.error(f"Monitor error: {e}")
            
            await asyncio.sleep(1)  # Check every second
    
    async def _update_level(self):
        """Update current backpressure level from metrics."""
        lag = await self.metrics.get_consumer_lag()
        depth = await self.metrics.get_queue_depth()
        
        old_level = self.current_level
        
        # Determine new level
        if lag >= self.config.lag_emergency:
            self.current_level = BackpressureLevel.EMERGENCY
        elif lag >= self.config.lag_critical or depth >= self.config.queue_depth_critical:
            self.current_level = BackpressureLevel.CRITICAL
        elif lag >= self.config.lag_warning or depth >= self.config.queue_depth_warning:
            self.current_level = BackpressureLevel.WARNING
        else:
            self.current_level = BackpressureLevel.NONE
        
        # Log level changes
        if self.current_level != old_level:
            logger.warning(
                f"Backpressure level changed: {old_level.name} → {self.current_level.name} "
                f"(lag={lag:,}, depth={depth:,})"
            )
    
    async def _adjust_loop(self):
        """Background loop to adjust rate limits and scaling."""
        while True:
            try:
                await self._adjust_rate_limit()
                await self._adjust_scaling()
            except Exception as e:
                logger.error(f"Adjust error: {e}")
            
            await asyncio.sleep(10)  # Adjust every 10 seconds
    
    async def _adjust_rate_limit(self):
        """Adjust rate limit based on current level."""
        if not self.config.enable_rate_limiting:
            return
        
        old_rate = self.current_rate_limit
        
        if self.current_level == BackpressureLevel.EMERGENCY:
            # Minimum rate
            self.current_rate_limit = self.config.min_rate_limit
            
        elif self.current_level == BackpressureLevel.CRITICAL:
            # Reduce by 50%
            self.current_rate_limit = max(
                self.config.base_rate_limit * 0.5,
                self.config.min_rate_limit
            )
            
        elif self.current_level == BackpressureLevel.WARNING:
            # Reduce by 25%
            self.current_rate_limit = max(
                self.config.base_rate_limit * 0.75,
                self.config.min_rate_limit
            )
            
        else:
            # Normal rate
            self.current_rate_limit = self.config.base_rate_limit
        
        if abs(self.current_rate_limit - old_rate) > 100:
            logger.info(f"Rate limit adjusted: {old_rate:.0f} → {self.current_rate_limit:.0f}/sec")
    
    async def _adjust_scaling(self):
        """Adjust consumer count based on lag."""
        if not self.config.enable_auto_scaling or not self.scaler:
            return
        
        # Check cooldown
        if self.last_scale_time:
            cooldown = datetime.now(timezone.utc) - self.last_scale_time
            if cooldown.total_seconds() < self.config.scale_cooldown_seconds:
                return
        
        lag = await self.metrics.get_consumer_lag()
        current_count = await self.scaler.get_replica_count()
        
        desired_count = current_count
        
        if lag > self.config.scale_up_threshold and current_count < self.config.max_consumers:
            # Scale up
            desired_count = min(current_count + 2, self.config.max_consumers)
            
        elif lag < self.config.scale_down_threshold and current_count > self.config.min_consumers:
            # Scale down
            desired_count = max(current_count - 1, self.config.min_consumers)
        
        if desired_count != current_count:
            logger.info(f"Scaling consumers: {current_count} → {desired_count} (lag={lag:,})")
            await self.scaler.scale_to(desired_count)
            self.last_scale_time = datetime.now(timezone.utc)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get current backpressure statistics."""
        total = self.messages_accepted + self.messages_shed + self.messages_rate_limited
        
        return {
            "level": self.current_level.name,
            "rate_limit": self.current_rate_limit,
            "accepted": self.messages_accepted,
            "shed": self.messages_shed,
            "rate_limited": self.messages_rate_limited,
            "shed_rate": (self.messages_shed / total * 100) if total > 0 else 0,
            "limited_rate": (self.messages_rate_limited / total * 100) if total > 0 else 0,
        }


# =============================================================================
# Priority-Aware Producer
# =============================================================================

class BackpressureAwareProducer:
    """
    Producer that respects backpressure signals.
    """
    
    def __init__(
        self,
        backpressure_manager: BackpressureManager,
        kafka_producer,
        topic: str
    ):
        self.bp_manager = backpressure_manager
        self.kafka = kafka_producer
        self.topic = topic
    
    async def send(
        self,
        message: Dict[str, Any],
        priority: Priority = Priority.MEDIUM,
        timeout: float = 10.0
    ) -> bool:
        """
        Send a message, respecting backpressure.
        
        Returns True if sent, False if rejected due to backpressure.
        """
        message['priority'] = priority.value
        
        # Check if we should accept this message
        if not await self.bp_manager.should_accept(message):
            return False
        
        # Send to Kafka
        try:
            await self.kafka.send_and_wait(self.topic, message)
            return True
        except Exception as e:
            logger.error(f"Send failed: {e}")
            return False
    
    async def send_critical(self, message: Dict[str, Any]) -> bool:
        """Send a critical message (never shed)."""
        return await self.send(message, Priority.CRITICAL)
    
    async def send_high(self, message: Dict[str, Any]) -> bool:
        """Send a high-priority message."""
        return await self.send(message, Priority.HIGH)
    
    async def send_low(self, message: Dict[str, Any]) -> bool:
        """Send a low-priority message (first to be shed)."""
        return await self.send(message, Priority.LOW)

# Consumer with Backpressure-Aware Processing

from aiokafka import TopicPartition


class BackpressureAwareConsumer:
    """
    Kafka consumer that handles its own backpressure.
    
    Features:
    - Monitors processing rate and queue depth
    - Pauses consumption when falling behind
    - Resumes when caught up
    - Reports metrics for external monitoring
    """
    
    def __init__(
        self,
        kafka_consumer,
        processor: Callable,
        max_batch_size: int = 100,
        max_processing_time: float = 30.0,  # seconds per batch
        pause_threshold: int = 10_000,       # Pause if internal queue > this
        resume_threshold: int = 1_000        # Resume if internal queue < this
    ):
        self.kafka = kafka_consumer
        self.processor = processor
        self.max_batch_size = max_batch_size
        self.max_processing_time = max_processing_time
        self.pause_threshold = pause_threshold
        self.resume_threshold = resume_threshold
        
        self._internal_queue: asyncio.Queue = asyncio.Queue()
        self._paused = False
        self._running = False
        
        # Metrics
        self.messages_processed = 0
        self.batches_processed = 0
        self.processing_time_total = 0.0
    
    async def start(self):
        """Start consumer loops."""
        self._running = True
        asyncio.create_task(self._fetch_loop())
        asyncio.create_task(self._process_loop())
        asyncio.create_task(self._backpressure_loop())
        logger.info("Backpressure-aware consumer started")
    
    async def stop(self):
        """Stop consumer gracefully."""
        self._running = False
        logger.info(
            f"Consumer stopped. Processed: {self.messages_processed:,}, "
            f"Batches: {self.batches_processed:,}, "
            f"Avg batch time: {self._avg_batch_time():.2f}s"
        )
    
    async def _fetch_loop(self):
        """Fetch messages from Kafka into internal queue."""
        while self._running:
            if self._paused:
                await asyncio.sleep(0.1)
                continue
            
            try:
                # Fetch batch from Kafka
                messages = await self.kafka.getmany(
                    timeout_ms=1000,
                    max_records=self.max_batch_size
                )
                
                for tp, msgs in messages.items():
                    for msg in msgs:
                        await self._internal_queue.put(msg)
                
            except Exception as e:
                logger.error(f"Fetch error: {e}")
                await asyncio.sleep(1)
    
    async def _process_loop(self):
        """Process messages from internal queue."""
        batch = []
        batch_start = time.monotonic()
        
        while self._running:
            try:
                # Get message with timeout
                try:
                    msg = await asyncio.wait_for(
                        self._internal_queue.get(),
                        timeout=0.1
                    )
                    batch.append(msg)
                except asyncio.TimeoutError:
                    pass
                
                # Process batch if full or time exceeded
                should_process = (
                    len(batch) >= self.max_batch_size or
                    (len(batch) > 0 and 
                     time.monotonic() - batch_start > self.max_processing_time)
                )
                
                if should_process and batch:
                    await self._process_batch(batch)
                    batch = []
                    batch_start = time.monotonic()
                    
            except Exception as e:
                logger.error(f"Process error: {e}")
    
    async def _process_batch(self, batch: List):
        """Process a batch of messages."""
        start = time.monotonic()
        
        try:
            await self.processor(batch)
            
            elapsed = time.monotonic() - start
            self.messages_processed += len(batch)
            self.batches_processed += 1
            self.processing_time_total += elapsed
            
            # Commit only what this batch processed. A bare commit() would
            # also commit messages still waiting in the internal queue.
            offsets = {
                TopicPartition(m.topic, m.partition): m.offset + 1
                for m in batch
            }
            await self.kafka.commit(offsets)
            
        except Exception as e:
            logger.error(f"Batch processing error: {e}")
            raise
    
    async def _backpressure_loop(self):
        """Monitor internal queue and pause/resume fetching."""
        while self._running:
            queue_size = self._internal_queue.qsize()
            
            if not self._paused and queue_size > self.pause_threshold:
                self._paused = True
                logger.warning(
                    f"Pausing fetch: internal queue at {queue_size:,} "
                    f"(threshold: {self.pause_threshold:,})"
                )
            
            elif self._paused and queue_size < self.resume_threshold:
                self._paused = False
                logger.info(
                    f"Resuming fetch: internal queue at {queue_size:,} "
                    f"(threshold: {self.resume_threshold:,})"
                )
            
            await asyncio.sleep(1)
    
    def _avg_batch_time(self) -> float:
        if self.batches_processed == 0:
            return 0.0
        return self.processing_time_total / self.batches_processed
    
    def get_metrics(self) -> Dict[str, Any]:
        return {
            "internal_queue_size": self._internal_queue.qsize(),
            "is_paused": self._paused,
            "messages_processed": self.messages_processed,
            "batches_processed": self.batches_processed,
            "avg_batch_time_seconds": self._avg_batch_time(),
        }

# Gradual Recovery Rate Limiter
from datetime import datetime, timezone


class GradualRecoveryRateLimiter:
    """Rate limiter that slowly restores rate after backpressure."""
    
    def __init__(
        self,
        base_rate: float,
        min_rate: float,
        recovery_steps: int = 5,
        step_duration_seconds: float = 60.0
    ):
        self.base_rate = base_rate
        self.min_rate = min_rate
        self.recovery_steps = recovery_steps
        self.step_duration = step_duration_seconds
        
        self.current_rate = base_rate
        self.recovery_start: Optional[datetime] = None
        self.in_recovery = False
    
    def reduce_rate(self, factor: float):
        """Reduce rate immediately during backpressure."""
        self.current_rate = max(self.base_rate * factor, self.min_rate)
        self.in_recovery = False
        self.recovery_start = None
    
    def start_recovery(self):
        """Start gradual rate recovery."""
        if not self.in_recovery:
            self.in_recovery = True
            self.recovery_start = datetime.now(timezone.utc)
    
    def get_current_rate(self) -> float:
        """Get current rate, accounting for gradual recovery."""
        if not self.in_recovery:
            return self.current_rate
        
        # Calculate recovery progress
        elapsed = (datetime.now(timezone.utc) - self.recovery_start).total_seconds()
        step = int(elapsed / self.step_duration)
        
        if step >= self.recovery_steps:
            # Fully recovered
            self.in_recovery = False
            self.current_rate = self.base_rate
            return self.base_rate
        
        # Interpolate between min and base
        progress = step / self.recovery_steps
        rate_range = self.base_rate - self.current_rate
        
        return self.current_rate + (rate_range * progress)