How to Run a Kafka Consumer Alongside Your App

The first Kafka consumer in most codebases is written in an afternoon and deployed inside the web app, because that is the deployment that already exists. It works. It keeps working for months. Then one Tuesday somebody ships four releases in an hour, and every release restarts the process, and every restart makes the consumer group stop the world and re-negotiate partition assignments — and the topic that was three seconds behind is now forty minutes behind and nobody noticed until a customer asked why their invoice never arrived.

I'm Ajay, I build PandaStack. This is the practical guide to running a Kafka consumer in the same machine as your web app: how to wire it up so it is actually correct, how to supervise it, and — the part most posts skip — the specific conditions under which you should stop doing this and give the consumer its own deployment.

Step 1: Decide whether co-location is right at all

Co-locating a consumer with the web process is a real pattern, not a hack, but it has a narrow window where it's the best option. Three things decide it.

Does the consumer block your request loop?

If your runtime is single-threaded and cooperatively scheduled — Node, or Python under a single-worker async server — a consumer that does 400ms of CPU-bound work per message is stealing 400ms from your HTTP handlers. Not concurrently. Instead of. A consumer that awaits I/O the whole time is mostly harmless; a consumer that parses a 5 MB payload and runs a regex over it is not. The honest test is to run your existing latency benchmark with the consumer active and see whether p99 moves. If it does, either move the work to a subprocess or split the consumer out entirely.

Do the two workloads want the same amount of hardware?

Web traffic and topic throughput are almost never correlated. Your web tier scales with concurrent users; the consumer scales with partition count and message rate. Co-location welds those together: you can only add consumer capacity by adding web capacity, and you can only add web capacity by adding consumers — which, since consumer count above partition count does nothing, is often just wasted money with a side of extra rebalances.

How often do you deploy?

This is the one that catches people, and it is worth being precise about. Kafka's consumer group protocol assigns partitions to members. When a member joins or leaves, the group rebalances. With the classic eager assignor, that means every member revokes every partition and the whole group stops consuming until assignment completes. Cooperative sticky rebalancing (the default in recent clients) is far gentler — it only moves the partitions it needs to — but it still costs a round trip through the coordinator, and it still happens on every single instance restart.

Now notice what co-location does: it couples the consumer's lifecycle to your web deploy cadence. A team that ships ten times a day has just signed their consumer group up for ten rebalances a day, minimum, multiplied by the number of app instances. If your deploy strategy is blue-green — start the new instance, then stop the old one — you get two rebalances per instance per deploy, and briefly twice as many members as you expected.

Step 2: Install the client and configure TLS + SASL from env vars

Any platform that gives you a real Linux userspace can do this — a container, a VM, a Firecracker microVM. On PandaStack an app runs in its own microVM with a full Ubuntu userspace, so the install is just part of the build command. The Python client wraps librdkafka, which needs a C toolchain unless a wheel exists for your platform, so install the build deps defensively:

# Build command (non-login sh -c, so no shell profile is sourced).
apt-get update -qq \
  && apt-get install -y --no-install-recommends ca-certificates librdkafka-dev \
  && pip install --no-cache-dir -r requirements.txt

# requirements.txt
#   confluent-kafka==2.6.1
#   orjson==3.10.7

Now the config. Every managed Kafka — Confluent Cloud, MSK with IAM off, Redpanda, Aiven — hands you the same four facts: bootstrap servers, a security protocol, a SASL mechanism, and a username/password pair. Those belong in environment variables, and nowhere else. The one that people get wrong is `ssl.ca.location`: if you install `ca-certificates` the system bundle is usually found automatically, but if you are on a slim base image with no bundle at all, the client fails with a certificate verification error that reads like a network problem and isn't.

# config.py
import os


def consumer_config() -> dict:
    return {
        "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP_SERVERS"],
        "security.protocol": os.getenv("KAFKA_SECURITY_PROTOCOL", "SASL_SSL"),
        "sasl.mechanisms": os.getenv("KAFKA_SASL_MECHANISM", "SCRAM-SHA-512"),
        "sasl.username": os.environ["KAFKA_SASL_USERNAME"],
        "sasl.password": os.environ["KAFKA_SASL_PASSWORD"],

        # Group identity. Keep this stable across deploys -- changing it
        # silently creates a NEW group that starts from auto.offset.reset.
        "group.id": os.environ["KAFKA_GROUP_ID"],

        # We commit offsets ourselves, after the work is done.
        "enable.auto.commit": False,

        # Only used the very first time this group ever connects.
        "auto.offset.reset": "earliest",

        # Cooperative sticky: a joining member takes only the partitions
        # it needs instead of the whole group dropping everything.
        "partition.assignment.strategy": "cooperative-sticky",

        # How long the coordinator waits after we stop heartbeating before
        # declaring us dead and rebalancing without us. See step 5 -- a
        # clean close() beats waiting this out every time.
        "session.timeout.ms": 45000,
        "heartbeat.interval.ms": 3000,

        # The real deadline: if one poll-to-poll gap exceeds this, the broker
        # kicks us out mid-batch. Size it to your SLOWEST message, not your
        # median one.
        "max.poll.interval.ms": 300000,
    }

The Node equivalent is the same shape with different key names. `kafkajs` is the pragmatic default — pure JS, no native build step, which matters if your deploy pipeline is a plain `npm ci`.

// kafka.js
import { Kafka, logLevel } from "kafkajs";

export const kafka = new Kafka({
  clientId: process.env.KAFKA_CLIENT_ID ?? "web-app",
  brokers: process.env.KAFKA_BOOTSTRAP_SERVERS.split(","),
  ssl: true,
  sasl: {
    mechanism: "scram-sha-512",
    username: process.env.KAFKA_SASL_USERNAME,
    password: process.env.KAFKA_SASL_PASSWORD,
  },
  logLevel: logLevel.WARN,
});

export const consumer = kafka.consumer({
  groupId: process.env.KAFKA_GROUP_ID,
  // kafkajs' equivalent of max.poll.interval.ms.
  sessionTimeout: 45000,
  heartbeatInterval: 3000,
});

Step 3: Commit after the work, be idempotent, and have somewhere to put poison

Three properties make a consumer correct, and they are all in tension with the code that is easiest to write.

1. Commit the offset AFTER the side effect, not before. Committing first gives you at-most-once delivery, which means a crash between commit and work silently loses a message with no error anywhere. Committing after gives you at-least-once, which means a crash between work and commit reprocesses one message. Choose the one that duplicates, because duplicates are a problem you can solve in code.
2. Which is why the handler must be idempotent. At-least-once is only safe if running the same message twice is harmless. Derive an idempotency key from something stable in the payload — an event id, or topic-partition-offset if the producer gives you nothing better — and make the write conditional on it.
3. A message that will never succeed must leave the partition. One un-parseable payload with an infinite retry loop stops that partition forever, and Kafka has no per-message dead-letter machinery. You build it: after N attempts, produce the message to a `<topic>.dlq` topic with the error attached, commit the offset, and move on.

# consumer.py -- commit after work, idempotent handler, DLQ for poison.
import logging
import os
import signal
import sys

import orjson
from confluent_kafka import Consumer, Producer, KafkaException

from config import consumer_config

log = logging.getLogger("consumer")
TOPIC = os.environ["KAFKA_TOPIC"]
DLQ_TOPIC = TOPIC + ".dlq"
MAX_ATTEMPTS = 3

running = True


def _stop(signum, _frame):
    # Just flip the flag. Doing real work in a signal handler is how you
    # end up with a half-closed consumer. See step 5.
    global running
    running = False
    log.info("signal %s received, draining", signum)


signal.signal(signal.SIGTERM, _stop)
signal.signal(signal.SIGINT, _stop)


def handle(event: dict, db) -> None:
    """Idempotent by construction: the unique index does the work."""
    db.execute(
        """
        INSERT INTO processed_events (event_id, kind, payload)
        VALUES (%s, %s, %s)
        ON CONFLICT (event_id) DO NOTHING
        """,
        (event["id"], event["kind"], orjson.dumps(event)),
    )


def main(db) -> int:
    consumer = Consumer(consumer_config())
    dlq = Producer({k: v for k, v in consumer_config().items()
                    if not k.startswith(("group.", "enable.auto", "auto.offset",
                                         "session.", "heartbeat.", "max.poll",
                                         "partition."))})
    consumer.subscribe([TOPIC])

    try:
        while running:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())

            for attempt in range(1, MAX_ATTEMPTS + 1):
                try:
                    handle(orjson.loads(msg.value()), db)
                    break
                except Exception as exc:  # noqa: BLE001
                    if attempt == MAX_ATTEMPTS:
                        log.exception("poison message -> DLQ")
                        dlq.produce(
                            DLQ_TOPIC,
                            key=msg.key(),
                            value=msg.value(),
                            headers=[
                                ("error", str(exc)[:512].encode()),
                                ("origin", f"{msg.topic()}/{msg.partition()}/{msg.offset()}".encode()),
                            ],
                        )
                        dlq.flush(5.0)
                    else:
                        log.warning("attempt %d failed: %s", attempt, exc)

            # Only now. The offset means "everything up to here is DONE",
            # including the ones we gave up on and parked in the DLQ.
            consumer.commit(message=msg, asynchronous=False)
    finally:
        # This is the important line. Step 5 explains why.
        consumer.close()

    return 0


if __name__ == "__main__":
    sys.exit(main(db=...))

Two details worth calling out. `commit(asynchronous=False)` is a synchronous round trip and it is slower — for high-throughput topics, commit every N messages instead and accept that a crash replays up to N. And the DLQ produce is flushed before the commit, because a DLQ write you never confirmed followed by a commit is exactly the data loss you were trying to avoid.

Step 4: Supervise the consumer next to the web process

You want the consumer running as a background child under a restart loop, and your web server in the foreground as PID 1's direct child, so the platform's health checks and log capture still watch the thing that serves traffic. This is the same shape as supervising a co-located Redis, and it fails in the same way if you get the last line wrong.

#!/bin/sh
# start.sh -- consumer supervised in the background, web app in the foreground.
set -e

# Restart the consumer forever, with a backoff so a crash loop against an
# unreachable broker doesn't spin the CPU.
( backoff=1
  while true; do
    python -m consumer || true
    echo "consumer exited (rc=$?), restarting in ${backoff}s" >&2
    sleep "$backoff"
    backoff=$(( backoff < 30 ? backoff * 2 : 30 ))
  done ) &
CONSUMER_SUPERVISOR=$!

# Forward SIGTERM to the consumer so a deploy gives it a chance to close
# the group membership cleanly instead of being killed outright.
trap 'kill -TERM "$CONSUMER_SUPERVISOR" 2>/dev/null; sleep 3' TERM INT

exec gunicorn app:app --bind 0.0.0.0:"${PORT:-8080}" --workers 2

The `exec` on the last line is not decoration. Without it your web server is a grandchild of the shell, the platform sends SIGTERM to the shell, and your app never hears about the shutdown at all. With it, the web process replaces the shell and gets signals directly.

On PandaStack both processes write to the same place: app stdout and stderr land in `/var/log/pandastack-app.log` inside the VM. That is convenient and also a small trap — interleaved web and consumer logs are hard to grep six months later, so prefix your consumer's log lines or ship them with a `component` field from the start.

Step 5: Shut down gracefully so the group rebalances fast

Here is the mechanism, because it explains an outage shape that looks unrelated to deploys. When a consumer process is killed without closing, it simply stops heartbeating. The group coordinator does not know it died — it only knows heartbeats stopped — so it waits out `session.timeout.ms` before declaring the member dead and reassigning its partitions. With the 45-second timeout above, that means those partitions are consumed by nobody for up to 45 seconds after every ungraceful restart.

A clean `consumer.close()` sends an explicit LeaveGroup request. The coordinator reacts immediately. The gap shrinks from tens of seconds to the time the rebalance itself takes. Same deploy, same partitions, one function call of difference.

• Handle SIGTERM by setting a flag, never by doing work inside the handler. Let the poll loop notice and exit through the normal path so the `finally` block runs.
• Close the consumer in a `finally`, so it also happens when the loop exits on an exception rather than a signal.
• Finish or abandon the in-flight message before closing, and commit what you completed. An uncommitted message will be redelivered — which is fine, because step 3 made the handler idempotent.
• Give yourself enough shutdown grace. If the platform's SIGTERM-to-SIGKILL window is shorter than your slowest message, graceful shutdown is theatre. Either shorten the work or lengthen the window.

Step 6: Watch consumer lag, and mostly only consumer lag

Consumer lag is the only Kafka metric that has ever woken anyone up, and it is the one most dashboards omit in favour of a wall of broker-side charts nobody reads. Lag is the difference between the newest offset in a partition and the offset your group has committed. It answers the only question that matters: how far behind reality is this consumer, and is that number growing?

# The one command worth putting in a runbook.
kafka-consumer-groups.sh \
  --bootstrap-server "$KAFKA_BOOTSTRAP_SERVERS" \
  --command-config client.properties \
  --describe --group "$KAFKA_GROUP_ID"

# GROUP   TOPIC     PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# orders  invoices  0          918_442         918_447         5    web-app-7f3c
# orders  invoices  1          912_005         961_880         49875  -
#
# Read it column by column:
#   LAG growing on ONE partition   -> a hot key, or a slow message stuck in a
#                                     retry loop on that partition only.
#   LAG growing on ALL partitions  -> you are simply under-provisioned, or the
#                                     handler got slower in the last release.
#   CONSUMER-ID empty ("-")        -> nobody owns that partition right now.
#                                     Mid-rebalance, or you have fewer live
#                                     members than partitions.

Alert on the derivative, not the value. Lag of 50,000 during a backfill is healthy; lag of 400 that has been climbing steadily for twenty minutes is an incident. The alert I actually want is "lag has increased in each of the last N intervals", because that catches a consumer processing slower than the producer regardless of the absolute number.

The two supporting metrics worth having: rebalance rate (a group that rebalances constantly is thrashing, usually from `max.poll.interval.ms` evictions) and DLQ depth (a DLQ that suddenly grows means a producer changed a schema and nobody told you).

Step 7: The exit conditions — when the consumer needs its own deployment

Decide these in advance, because in the middle of an incident everything looks like a reason to change everything.

• Deploy frequency outgrew the lag budget. When routine web releases are visibly moving your lag graph, the consumer's lifecycle needs to stop being tied to the web tier's.
• Consumer work is measurably hurting request latency. If p99 moves when the consumer is busy, they are competing for the same CPU and one of them should leave.
• You need more consumers than web instances, or fewer. The moment the right number of each diverges, co-location is costing you money in one direction or throughput in the other.
• Partition count went up. More partitions is the standard answer to lag, and it only helps if you can actually run more consumer members — which, under co-location, means running more web instances you do not need.
• The handler needs different resources. A consumer that wants 8 GiB for batch aggregation should not be forcing your web tier onto 8 GiB machines.
• Different on-call. When queue processing has its own owner and its own pager, it should have its own deployment. Shared processes make blast radius arguments impossible to win.

The migration itself is genuinely easy, which is the good news: same repository, same build, different start command, same environment variables, same `group.id`. Kafka doesn't care where a group member runs. Deploy the new consumer-only app, then remove the consumer from the web app's start command, and the group rebalances once and carries on.

The honest caveat: consumers and scale-to-zero pull in opposite directions

A co-located consumer lives and dies with the machine. That is fine on always-on infrastructure and it is a real design tension on anything that sleeps idle apps, including PandaStack. A web app can sleep because an inbound HTTP request wakes it — the platform holds the request while the machine comes back. A Kafka consumer has no inbound request. It discovers work by polling the broker, so it either polls forever (which counts as activity and keeps the machine awake, which is the whole point) or it sleeps and nobody is consuming when messages arrive.

There is no clever resolution to that. On PandaStack, an app with a live consumer keeps its VM warm, and that is a cost decision you should make on purpose rather than discover on an invoice. Three ways to make it deliberately:

1. Accept the warm VM. If the topic needs sub-minute processing, you are buying always-on capacity, and that is the correct purchase. Size the machine for the consumer's actual working set rather than leaving the default.
2. Batch instead of stream. For latency-tolerant work, run a scheduled job that starts, drains the topic until the poll returns empty, commits, and exits. You trade freshness for a machine that is idle most of the day. Snapshot-restore create on PandaStack has a p50 of 179ms and a p99 around 203ms, so starting a fresh machine per drain is cheap enough that a five-minute schedule is entirely reasonable.
3. Push instead of poll. If you control the producer side, an HTTP delivery — a webhook, or a bridge that reads Kafka and POSTs — turns job delivery into a request the sleeping app can wake on. This is the only option that gets you both low latency and an idle-to-zero app, and it costs you a component to run.

Want to test the shape before committing? Spin up a throwaway machine and run the consumer against a staging topic:

from pandastack import Sandbox

sbx = Sandbox.create(template="base", ttl_seconds=600)

sbx.exec("pip install -q confluent-kafka orjson")
sbx.exec("python -m consumer --once", env={
    "KAFKA_BOOTSTRAP_SERVERS": "...",
    "KAFKA_GROUP_ID": "staging-drain-probe",
})

The summary

Run the consumer next to your web app when you deploy infrequently, the handler is I/O-bound, and both workloads want the same hardware. Configure TLS and SASL from environment variables, turn off auto-commit, and commit after the side effect so you get at-least-once with an idempotent handler behind it. Give poison messages a dead-letter topic so one bad payload can't stall a partition. Supervise the consumer in a restart loop and `exec` the web server so signals land where they should. Close the consumer on SIGTERM so the group rebalances in seconds instead of waiting out the session timeout. Alert on lag increasing, not lag existing. And split the consumer into its own deployment the day deploy frequency, latency, or scaling needs diverge — which is a fifteen-minute change if you set it up this way from the start.