Backpressure Mechanisms

Concepts covered: Stream Processing

Backpressure is a stage telling the stage before it how much it can accept. With it, excess load waits in a place with a known bound or is refused at the edge. Without it, excess load piles up wherever nothing is counting, usually a process's heap, until the process dies and takes its in-flight work with it.

Where excess load can go

A pull-based queue provides buffering without extra design. Consumers fetch when they have room, so the backlog sits in the queue's partitions on disk, the lag is the pressure signal, and the retention period is the bound. Push-based links, where the sender decides when data moves, need an explicit signal instead: the receiver grants credits for a number of records, and the sender stops when the credits run out.

The unbounded buffer inside a consumer

The pull model protects you only up to the consumer's front door. A common consumer design runs a fetch loop that pulls records into an in-memory list and a pool of threads that enrich from it. If that list has no limit, the fetch loop keeps pulling at the partition's arrival rate no matter how fast enrichment drains it.

Run the numbers at doubled volume. Each of the 24 partitions receives around 3,333 events per second. Now the profile service degrades from 4 ms to 40 ms per call, and each consumer's 10 threads finish only 250 events per second. The list grows by around 3,083 events, or 3 MB, every second. A 2 GB heap fills in about 11 minutes. The consumer crashes, its uncommitted records are fetched again by whichever consumer inherits the partition, and that consumer fills its heap the same way. The pipeline now produces crashes and duplicate writes on top of the lag.

import queue

buffer = queue.Queue(maxsize=10_000)


def fetch_loop(consumer):
    paused = False
    while True:
        for record in consumer.poll(max_records=500):
            buffer.put(record)
        if not paused and buffer.qsize() >= 9_500:
            consumer.pause(consumer.assignment())
            paused = True
        elif paused and buffer.qsize() <= 5_000:
            consumer.resume(consumer.assignment())
            paused = False


def enrich_loop(profile_client, writer):
    while True:
        record = buffer.get()
        profile = profile_client.lookup(record["driver_id"])
        writer.write({**record, "driver_tier": profile["tier"]})
        buffer.task_done()

At 9,500 waiting records, fetch_loop pauses its partitions but keeps calling poll, which returns nothing yet proves the consumer alive. A loop that blocked would stop polling, and past the group's maximum poll interval the consumer is ejected and its uncommitted records replayed elsewhere. Unread records stay durable in the partition, memory stays near 10 MB per consumer, and the slowdown becomes lag. Commit offsets only after the writer accepts a record. With 10 threads finishing out of order, commit the lowest offset not yet written, so a crash replays work and loses none.

Choosing a mechanism per event class

The trip-events stream carries 2 kinds of event. Location pings, sent every few seconds during a trip, are 60% of volume. Lifecycle events (requested, accepted, started, completed, fare finalized) are the other 40%. At doubled peak that is 48,000 pings and 32,000 lifecycle events per second, and the 2 kinds deserve different treatment.

Location pings

• A 50% sample still traces the route on a map.
• Value decays fast: a position from 30 minutes ago is useless.
• Shed at the gateway, before the event costs a lookup.
• Shedding half removes 24,000 events per second at doubled peak.

Lifecycle events

• Every event moves money: a lost fare event is a wrong payout.
• Value survives delay: an earnings row 20 minutes late is still correct.
• Buffer in the queue and never shed.
• Order matters per trip, which shapes the partition key.

Shedding half the pings brings doubled input to 56,000 events per second against 60,000 of consumer capacity. The consumers run at 93% of capacity, so the lag stops growing, but there is almost no room to recover from any blip. Shedding buys the time it takes to add partitions and consumers.

It falls behind. At 8 ms, a thread handles 125 events per second, a consumer handles 1,250, and 24 consumers handle 30,000. Input after shedding is 56,000, so lag grows by 26,000 events per second. Shedding fixes a load problem. A latency problem in a dependency needs more concurrency per consumer, fewer calls per event, or a decision to stop calling the dependency on the hot path.

Do

• Bound every in-memory buffer by count or bytes
• Commit offsets after the write succeeds
• Decide shed or buffer per event class, in advance
• Shed at the earliest stage, before work is spent

Avoid

• Let a fetch loop outrun the threads that drain it
• Shed events that drive money or state
• Retry throttled calls on a fixed schedule
• Treat a durable queue as unlimited: retention is the bound

About this lesson

This section is part of the Backpressure and Scaling: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.