Pausing and Resuming Kafka Consumers for Slow Sinks #

When the sink slows down, a Kafka consumer has three options and only one of them is correct. It can keep consuming and buffer in memory, which relocates the problem and eventually kills the process. It can stop calling poll(), which the group coordinator interprets as death and answers with a rebalance that stalls every other member too. Or it can pause its partitions — continuing to poll and heartbeat while fetching nothing — and resume when the sink recovers. This guide implements the third, as the transport-layer counterpart to the bounded in-process queue within backpressure handling in Python IoT streams.

Why Pause Exists #

Kafka’s consumer protocol has two independent liveness mechanisms and it is essential to know which one you are about to violate.

session.timeout.ms is about heartbeats, which the client sends from a background thread. A consumer that is busy processing still heartbeats, so a long processing pause does not trip this.

max.poll.interval.ms is about progress: the time between successive poll() calls on the application thread. It exists precisely to catch a consumer that is alive but stuck — and a consumer sleeping until its database comes back is, from the coordinator’s perspective, indistinguishable from one deadlocked in a lock. Exceed it and the member is removed from the group, its partitions are reassigned, and every other member stops while that happens.

pause() threads the needle. It tells the client “stop fetching records for these partitions”, so poll() returns promptly with nothing. The application keeps polling at its normal cadence, max.poll.interval.ms is never approached, and the group is undisturbed. Meanwhile the broker’s retention is doing the buffering, which is what it is for.

Production-Ready Implementation #

# python 3.11 · confluent-kafka==2.4.0 · prometheus-client==0.20.0
from __future__ import annotations

import logging
import time

from confluent_kafka import Consumer, KafkaException
from prometheus_client import Counter, Gauge

log = logging.getLogger(__name__)

PAUSED = Gauge("consumer_partitions_paused", "Partitions currently paused")
PAUSE_EVENTS = Counter("consumer_pause_events_total", "Times the consumer paused", ["reason"])


class SinkAwareConsumer:
    """Consumer that parks its partitions while the sink is unavailable.

    Pausing keeps poll() flowing, so the group coordinator sees a healthy member
    and no rebalance is triggered — the failure stays local to this consumer.
    """

    def __init__(self, consumer: Consumer, sink, *, backoff_s: float = 1.0,
                 max_backoff_s: float = 60.0) -> None:
        self.consumer = consumer
        self.sink = sink
        self.backoff_s = backoff_s
        self.max_backoff_s = max_backoff_s
        self._paused = False
        self._retry_at = 0.0
        self._delay = backoff_s

    def _pause(self, reason: str) -> None:
        if self._paused:
            return
        assignment = self.consumer.assignment()
        self.consumer.pause(assignment)
        self._paused = True
        self._retry_at = time.monotonic() + self._delay
        PAUSED.set(len(assignment))
        PAUSE_EVENTS.labels(reason=reason).inc()
        log.warning("paused %d partition(s) for %.1fs: %s", len(assignment), self._delay, reason)

    def _resume(self) -> None:
        assignment = self.consumer.assignment()
        self.consumer.resume(assignment)
        self._paused = False
        self._delay = self.backoff_s          # reset the backoff on success
        PAUSED.set(0)
        log.info("resumed %d partition(s)", len(assignment))

    def run(self) -> None:
        while True:
            # poll() ALWAYS runs, paused or not — this is what keeps the group healthy
            msg = self.consumer.poll(timeout=1.0)

            if self._paused:
                if time.monotonic() >= self._retry_at:
                    if self.sink.healthy():
                        self._resume()
                    else:
                        self._delay = min(self.max_backoff_s, self._delay * 2)
                        self._retry_at = time.monotonic() + self._delay
                continue

            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())

            try:
                self.sink.write_batch([decode(msg.value())])
            except SinkUnavailable as exc:
                self._pause(reason=type(exc).__name__)
                continue                      # do NOT commit: the record will be redelivered

            self.consumer.store_offsets(msg)  # commit only after a successful write

Two lines carry the correctness of the whole class. poll() is called unconditionally at the top of the loop, including while paused — remove that and the pause becomes the sleep it was meant to replace. And store_offsets runs only after the sink write succeeds, so a record processed during a failure is redelivered rather than silently skipped, which is the same at-least-once contract as the Faust agent pattern.

The sink’s health check should be cheap and honest:

class PostgisSink:
    def healthy(self) -> bool:
        """A real round trip, not a cached flag — the point is to detect recovery."""
        try:
            with self.conn.cursor() as cur:
                cur.execute("SELECT 1")
            return True
        except Exception:
            return False

Parameter Tuning Guide #

The fourth row is the one to internalise: pausing on a per-record data error stops the whole partition because of one malformed reading. Route that row to a dead-letter path and keep going.

Verification and Testing #

# python 3.11 · pytest==8.2.0
class FlappingSink:
    """Unavailable for the first `outage` calls, healthy afterwards."""

    def __init__(self, outage: int = 3):
        self.remaining, self.written = outage, []

    def healthy(self) -> bool:
        return self.remaining <= 0

    def write_batch(self, rows):
        if self.remaining > 0:
            self.remaining -= 1
            raise SinkUnavailable("sink is down")
        self.written.extend(rows)


def test_consumer_keeps_polling_while_paused(fake_consumer):
    """The whole point: poll() must be called during the outage, or the group rebalances."""
    sink = FlappingSink(outage=3)
    c = SinkAwareConsumer(fake_consumer, sink, backoff_s=0.01)
    run_for(c, seconds=0.5)

    assert fake_consumer.poll_calls > 10           # polling continued throughout
    assert fake_consumer.paused_at_least_once
    assert sink.written                            # and it recovered


def test_offsets_are_not_stored_for_a_failed_write(fake_consumer):
    sink = FlappingSink(outage=1)
    c = SinkAwareConsumer(fake_consumer, sink, backoff_s=0.01)
    run_for(c, seconds=0.2)
    assert fake_consumer.stored_offsets == fake_consumer.successful_writes

The operational check is simpler and more convincing: stop the database for sixty seconds in staging and watch two things. The consumer group must not rebalance (no membership change in the broker logs), and consumer lag must rise and then fall back to baseline without intervention. If a rebalance appears, something in the loop is skipping poll().

Gotchas #

Pausing without resuming. If the resume path is inside an exception handler that never runs, the consumer polls forever and processes nothing — a stall that looks perfectly healthy from the outside because the member is alive and lag rises smoothly. Export the paused gauge and alert on it.

consumer.assignment() returning empty during a rebalance. Pausing an empty list silently does nothing, and the consumer keeps fetching into a failing sink. Re-apply the pause in the on_assign callback if the paused flag is set.

Auto-commit left enabled. With enable.auto.commit=true, offsets advance on a timer regardless of whether the sink write succeeded, so everything processed during the outage is marked done. This turns a graceful pause into silent data loss.

A health check that hits a cached connection object. Checking conn.closed tells you what the client believes, not what the server is doing. Issue a real query with a short timeout.

FAQ #

Related #

• Backpressure Handling in Python IoT Streams — the backpressure stage this belongs to
• Bounding asyncio Queues for Sensor Consumers — the in-process bound that decides when to pause
• Managing Consumer Group Rebalancing for Sensor Topics — the failure this technique exists to avoid