Backpressure Control

What Is Backpressure?

Backpressure occurs when a downstream system cannot keep up with the rate of incoming work. Without backpressure control, buffers fill up, memory is exhausted, and the system crashes. Backpressure is the signal that tells producers to slow down.

Causes of Backpressure

• Traffic spikes — Sudden surge in requests exceeds processing capacity.
• Slow consumers — A consumer processing a batch is slower than the producer.
• Dependency latency — A downstream service is slow, blocking the consumer.
• Resource exhaustion — CPU, memory, or disk I/O limits are reached.
• Partition imbalance — Some partitions have more data than others.

Bounded Buffers

Bounded buffers have a maximum size. When the buffer is full, the producer is blocked or new messages are rejected. This provides natural backpressure — the producer cannot outpace the consumer beyond the buffer size.

from queue import Queue, Full

# Bounded buffer: max 1000 messages
buffer = Queue(maxsize=1000)

def enqueue_message(message):
    try:
        buffer.put_nowait(message)  # Non-blocking
    except Full:
        # Backpressure: buffer is full
        log.warning("Buffer full, shedding load")
        metrics.increment("backpressure.messages_dropped")
        raise ServiceUnavailable("System overloaded")

Consumer Lag

Consumer lag is the difference between the latest message in a partition and the consumer's current position. High lag means messages are delayed. Monitor lag and alert when it exceeds thresholds. Lag can indicate slow consumers or insufficient capacity.

from kafka import KafkaConsumer
from prometheus_client import Gauge

lag_gauge = Gauge('kafka_consumer_lag', 'Consumer group lag', ['topic', 'partition'])

consumer = KafkaConsumer('orders', group_id='processor')

while True:
    for tp in consumer.assignment():
        committed = consumer.committed(tp)
        end_offset = consumer.end_offsets([tp])[tp]
        lag = end_offset - (committed or 0)
        lag_gauge.labels(tp.topic, tp.partition).set(lag)
        
        if lag > 10000:
            log.warning(f"High lag on {tp.topic}[{tp.partition}]: {lag}")

Load Shedding

Load shedding deliberately drops or rejects requests when the system is overloaded. This protects the system from complete failure by sacrificing some requests. Shed low-priority work first, then medium-priority, then high-priority.

import time

def process_with_priority(message):
    priority = message.get('priority', 'medium')
    queue_depth = get_queue_depth()
    
    # Shed low priority when queue > 50% full
    if priority == 'low' and queue_depth > 5000:
        return shed("low_priority_shed")
    
    # Shed medium priority when queue > 80% full
    if priority == 'medium' and queue_depth > 8000:
        return shed("medium_priority_shed")
    
    # Never shed high priority
    return process(message)

Adaptive Rate Limiting

Adaptive rate limiting adjusts rate limits based on system health. When the system is under pressure (high CPU, high lag), reduce the allowed rate. When the system recovers, increase the rate. This provides automatic backpressure without manual intervention.