Forem: Myroslav Vivcharyk

Ordered Retries in Kafka: Why You Probably Shouldn't Build This

Myroslav Vivcharyk — Tue, 24 Mar 2026 15:10:00 +0000

I once spent three weeks in architectural meetings arguing that we should insert SQS between two Kafka topics.

I wasn't doing it because I liked SQS. I was doing it because the "clean" Kafka-native retry logic felt like a complexity bomb our team wasn't ready to defuse.

As engineers, our job isn't just to build clever systems for the next promotion cycle. Our job is to build systems that don't set the on-call engineer's life on fire at 3 AM.

In Part 1 and Part 2, we built a Rube Goldberg machine of distributed locks, compacted topics, and reference counting. It works. It maintains strict ordering during failures without blocking entire partitions.

Before you import my library or build your own, let's talk about the engineering hangover that comes after the party.

The Technical Costs

1. The Cold Start Penalty

Your service was stateless. Kubernetes could kill pods and spin up new ones in seconds. Not anymore.

Now your application carries state. That state lives in the lock topic. On startup, every pod must:

Connect to Kafka
Read the lock topic from the beginning
Rebuild the in-memory map before processing anything

With 100 active locks? Instant. With 500,000 lock records after a bad day — every lock and unlock is a separate record, your pod startup goes from seconds to minutes.

Here's where it gets ugly: your Horizontal Pod Autoscaler sees growing lag and spins up more pods. Those pods sit there "syncing state..." while the lag grows. The new pods can't help because they're still starting up. You've created a scaling death spiral.

Mitigation:

Aggressive retention on the lock topic (minutes, not days)
Tune max.poll.interval.ms to accommodate sync time
Use this pattern only for low-cardinality keys

2. The Memory Pressure

The lock map itself is small — a Go map with 100,000 string keys and integer counters fits in single-digit megabytes. That's not the problem.

The problem is what happens upstream. During a failure storm, every failed message triggers a lock write and a retry redirect. The lock topic and retry topic both grow rapidly. Your Kafka consumers buffer incoming records in memory before processing. The lock consumer, replaying a suddenly-huge topic during a restart, can spike memory well beyond steady-state expectations.

Combine this with Cost #1: your pod OOM-kills during state restore, restarts, tries again, OOM-kills again. Crash loop.

Mitigation:

Circuit breaker: if lock count exceeds threshold, fall back to stop-the-world
Set memory limits with headroom for failure storms, not steady-state

3. Topic Overhead

You wanted one topic: orders.

To make ordered retries work, you now operate four:

orders — main topic
orders-retry — the retry queue
orders-lock — the compacted state topic
orders-dlq — dead letters

200 topics across 50 services won't bring your Kafka cluster down — production clusters run thousands. The cost is operational: each topic needs provisioning, monitoring, retention policies, and alerting. Your SRE team will love you.

Mitigation:

Only use this pattern for topics that truly need it
Budget the operational overhead before committing

4. Zombie Lock Operations

In Part 2, we discussed zombie locks — keys locked forever because the retry message never made it. The fix was compensating transactions and careful ordering.

But "rare" still means "happens." When it does, you need to detect and clean it up.

Detection:

Alert on locks older than N hours
Compare lock topic keys against retry topic — orphans are zombies
Dashboard showing lock age distribution

Cleanup:

Manual tombstone producer (you'll need the CLI tool)
Automated reaper: background job that scans for stale locks and releases them
TTL simulation: treat locks older than X hours as expired

The reaper sounds easy until you realize: how do you know a lock is stale vs. just slow? A retry with exponential backoff might legitimately take hours. You need metadata — timestamps, attempt counts — in the lock record itself.

The question: Do you have this tooling built? Do you have observability for it? If not, you're one zombie lock away from a very long night.

The 3 AM Test

For me, this matters more than the technical drawbacks — even though they're intertwined.

Before building any complex system, try to write the runbook for it.

A downstream team calls you: "Why is Order-123 not updating?"

How many hops to find the answer?

Check the main consumer logs
See a "key locked, redirecting" message
Query the lock topic to confirm the lock exists
Query the retry topic to find the original failure
Realize the lock is 4 hours old and the retry message... isn't there

Now what? Do you have tooling to manually produce a tombstone? Do you know which partition that key hashes to? Can you do this at 3 AM, half-asleep, with Slack pinging?

Here's the kicker: the person debugging this at 3 AM might not be you. In many organizations, the on-call rotation is a separate team — engineers who've never seen your code, who are juggling alerts from a dozen services. Can they navigate your distributed lock registry with a runbook, or are they going to escalate and wake you up anyway?

And there's another problem: Kafka doesn't have key-based lookup. You can't just say "show me the lock for Order-123." To find a specific key, you either scan the entire topic or build custom tooling that maintains an index. Does your company have that tooling? Will you build and maintain it? How will this system behave during a global failure when hundreds of keys are locked simultaneously?

If you can't visualize this "penalty box" of locked keys — which keys are locked, for how long, and why — you haven't built a resilient system. You've built a black box.

The rule: If you can't write the runbook, you can't operate the system.

Observability

The alerts that matter:

acquired - released growing over time → locks aren't clearing, potential zombies
state_restore.duration approaching max.poll.interval.ms → rebalance loop incoming (see Part 2)
dlq.enqueued > 0 → manual intervention required
redirected spike without corresponding retry.processed increase → retries aren't succeeding

Without this, you're flying blind. Build the dashboards before you ship the feature.

Before You Build This

Before reaching for distributed locks, exhaust the simpler options.

Option 1: Make It Commutative

I know — in an existing system, this is easy to say and almost impossible to do. But if you're building something new, ask: do you really need strict ordering?

Sometimes you can redesign the problem away. Store deltas instead of absolute values. Use event sourcing with snapshots and recalculate on late arrivals. Add idempotency keys so duplicates and reordering become harmless. These aren't always possible. But when they are, they eliminate the problem entirely rather than managing it.

Option 2: External State Store (Redis/Dragonfly)

Most of you reading this probably thought: "Why not just use Redis and skip all this Kafka-native complexity?"

Fair point. Redis gives you:

Native TTL — zombie locks expire automatically, no reaper needed
Instant startup — no cold start problem, no state to rebuild
Fast lookups — no consumer lag to worry about, O(1) key access
Familiar tooling — your team probably already knows how to operate Redis

But this isn't free either. You're adding a new infrastructure dependency. Your Kafka consumer now fails if Redis is down. Network latency on every lock check adds up. And now you're debugging two systems instead of one.

A word on Redlock: if you're considering distributed locking with Redis, read Martin Kleppmann's analysis first. Distributed locking is hard regardless of the tool. Redis doesn't make the fundamental problem easier — it makes the infrastructure easier.

My take: If I were starting fresh today and needed ordered retries with blocking semantics, I'd reach for Redis/Dragonfly before building the Kafka-native solution we covered in Parts 1 and 2. Redis doesn't eliminate complexity — it moves it somewhere your team probably already understands.

Option 3: Accept the Trade-off

Sometimes the "stop-the-world" approach from Part 1 is actually fine.

If your topic has low throughput — blocking the partition during a retry might be acceptable. The complexity of distributed locks only pays off when throughput requirements make blocking unacceptable.

Run the numbers. If your worst-case retry takes 5 minutes and your partition processes 100 messages per hour, you're blocking 8 messages. Is that worth an operational burden?

When It's Worth It

After all that, there are cases where this pattern might be the right call:

Strict ordering is contractual — Downstream systems will break or charge you money if events arrive out of order. Not "might cause weird bugs" — actually break, with consequences.
Stop-the-world is unacceptable — Throughput requirements mean you can't block a partition for minutes. You've measured this, not assumed it.
You have the tooling for debugging and observability — Your team can see which keys are locked, for how long, and why. You can produce a tombstone manually. An SRE can follow the runbook at 3 AM without escalating.
Your team understands it — The next engineer can debug it without reading three blog posts. You've documented the architecture, the failure modes, and the recovery procedures.

If all four are true, go ahead. The kafka-resilience library handles the edge cases we covered in Part 2.

If any are false, reconsider.

The Tax

Engineering is the art of choosing which problems you want to have.

This solution solves the ordering problem. In exchange, it gives you the state management problem, the cold start problem, the memory pressure problem, and the 3 AM debugging problem.

The technical challenges are solvable. The question is whether you should be the one solving them — or whether a simpler solution, even an imperfect one, is the wiser choice for your team and your system.

Build the clever thing when you've exhausted the simple options. Not before.

Ordered Retries in Kafka: The Bugs You'll Find in Production

Myroslav Vivcharyk — Tue, 17 Mar 2026 15:10:00 +0000

In Part 1, we introduced a "middle ground" for Kafka retries: when a message fails, lock its key, send the message to a retry topic, and let the main partition continue processing other keys. If you need a recap, the Confluent blog covers the pattern.

Simple on a whiteboard. In production, things break in ways the diagrams don't show.

This article covers implementation gaps that tutorials skip. The code examples are from kafka-resilience — simplified for clarity, but the edge cases are real.

The Architecture Recap

The goal: when a message fails, block only that key. Other keys keep flowing.

To make this work across multiple consumer instances, we need three topics:

Main Topic — Your business events. The consumer checks if a key is locked before processing.
Retry Topic — Messages land here either because they failed or because a predecessor with the same key failed. A separate consumer processes them with backoff. When a message succeeds here, it releases the lock.
Lock Topic (compacted) — The shared lock registry. When Instance A locks "Order-123", it writes here. Instance B reads this topic to learn about locks it didn't create.

Each instance runs two consumer loops: one for business messages (main or retry), and one that continuously reads the lock topic to keep the local lock map in sync.

Here's how they connect:

Here's where it breaks.

Problem 1: The Dual-Write

When a message fails, you need to do two things:

Write a lock to the lock topic
Write the payload to the retry topic

If the first succeeds but the second fails, you've created a zombie lock — the key is locked forever, but there's no message in the retry queue to unlock it.

Solution A: Kafka Transactions

The obvious solution is Kafka transactions — wrap both writes in a transaction so they either both succeed or both fail.

Most modern clients support this, even outside the Java ecosystem. If your setup allows it, this is a valid approach.

Why I didn't use it: Kafka transactions aren't just a flag you flip. They require a transactional producer, which means a dedicated transaction coordinator on the broker side. Every transactional write follows an init → begin → produce → commit cycle — that's at least two extra round-trips per message. With acks=all across availability zones, each round-trip adds latency.

Under normal load, this overhead is manageable. During a failure storm — exactly when your retry system is under heavy load — it compounds. If 10,000 messages fail in a burst, you're running 10,000 transactions, each waiting for cross-AZ acknowledgments before committing. The retry system becomes the bottleneck at the worst possible moment.

Transactions are the correct solution if your broker topology can absorb the overhead. For high-throughput systems, there's a lighter alternative.

Solution B: Compensating Transaction

Instead, I used a compensating transaction: acquire the lock first, then try to write the payload. If the write fails, undo the lock by writing a release.

When processing fails, the Tracker calls redirect to lock the key and forward the message to the retry topic.

func (t *Tracker) redirect(ctx context.Context, msg Message) error {
    // Step 1: Acquire lock first
    if err := t.coordinator.Acquire(ctx, msg); err != nil {
        return fmt.Errorf("failed to acquire lock: %w", err)
    }

    // Step 2: Publish to retry topic
    if err := t.publisher.Send(ctx, t.retryTopic, msg); err != nil {
        // Step 3: Compensate on failure — undo the lock
        if releaseErr := t.coordinator.Release(ctx, msg); releaseErr != nil {
            t.logger.Error("compensation failed - potential zombie lock",
                "key", msg.Key(),
                "error", releaseErr)
        }
        return fmt.Errorf("failed to publish to retry: %w", err)
    }

    return nil
}

The trade-off: if the process crashes between acquiring the lock and completing the compensation, you get a zombie. Rare, but possible. For those edge cases, you'll need a background "reaper" to clean stale locks. We'll cover that in Part 3.

Whichever solution you choose, the remaining problems still apply.

Problem 2: The Reference Counting Gap

Most descriptions of this pattern treat the retry topic as a queue of failed messages. It's not. It also holds messages that never failed — they were redirected because a predecessor with the same key did. This distinction matters now.

Watch this sequence carefully:

T1: Event A (Key: 123) fails
    → Lock Key 123
    → Messages in retry: 1

T2: Event B (Key: 123) arrives, sees lock
    → Redirect to retry topic (not failed — just blocked)
    → Messages in retry: 2

T3: Event A retries successfully
    → Release lock... but HOW?

If "release" means "delete the lock," you've just broken ordering. Event B is still sitting in the retry topic. But the lock is gone. When Event C arrives for Key 123, it processes immediately — before B.

T4: Event C (Key: 123) arrives
    → Check lock: NOT LOCKED (wrong!)
    → Processes immediately
    → C processed before B

The lock isn't a boolean. It's a counter.

Each instance maintains a LocalCoordinator — an in-memory map counting pending retries per key. Later, we'll wrap this with a KafkaCoordinator that broadcasts state changes to other instances.

type LocalCoordinator struct {
    locks map[string]int  // key → reference count
    mu    sync.RWMutex
}

func (c *LocalCoordinator) Acquire(key string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.locks[key]++  // Increment
}

func (c *LocalCoordinator) Release(key string) {
    c.mu.Lock()
    defer c.mu.Unlock()

    c.locks[key]--
    if c.locks[key] <= 0 {
        delete(c.locks, key)  // Only free when count hits zero
    }
}

func (c *LocalCoordinator) IsLocked(key string) bool {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.locks[key] > 0
}

Now the sequence works correctly:

T1: Event A fails      → count = 1
T2: Event B redirects  → count = 2
T3: Event A succeeds   → count = 1 (still locked!)
T4: Event C arrives    → blocked (correct!)
T5: Event B succeeds   → count = 0 (now free)
T6: Event C proceeds   → processes in correct order

The main topic stays blocked for Key 123 until every pending message clears — both the ones that actually failed and the ones that were redirected to preserve order.

Problem 3: The Self-Consumption Trap

Reference counting works locally. But we're broadcasting every lock change to Kafka so other instances can learn about it. And Kafka doesn't filter — you'll also consume your own messages.

Instance A applies the lock twice: once when it acquires locally, once when it consumes its own broadcast. The reference count is now wrong. When the retry succeeds, the count drops to 1 instead of 0. The key stays locked forever.

Fix: tag every message with an origin ID and ignore your own.

const HeaderCoordinatorID = "x-coordinator-id"

func (c *KafkaCoordinator) Acquire(ctx context.Context, msg Message) error {
    c.local.Acquire(msg.Key())  // Apply locally first

    return c.producer.Send(ctx, c.lockTopic, &LockMessage{
        Key:   msg.Key(),
        Value: msg.Key(),
        Headers: map[string]string{
            HeaderCoordinatorID: c.instanceID,  // Tag with our ID
        },
    })
}

func (c *KafkaCoordinator) processLockMessage(msg Message) error {
    originID := msg.Headers().Get(HeaderCoordinatorID)

    // Skip our own messages
    if originID == c.instanceID {
        return nil
    }

    // Apply locks from other instances
    if msg.Value() != nil {
        return c.local.Acquire(msg.Key())
    }

    // Release locks for tombstones
    return c.local.Release(msg.Key())
}

Now each instance only applies foreign locks from the broadcast.

One detail: the instance ID must be stable across restarts. If a pod restarts and gets a new ID, it won't recognize its own pre-crash messages during replay and will double-count them. Use something deterministic — the consumer group member ID or a stable pod identifier — not a random UUID generated at startup.

Problem 4: The Compaction Collision

The lock topic is compacted — Kafka periodically deduplicates records, keeping only the latest value per key. Great for storage. Dangerous for counting.

Here's the trap: compaction doesn't happen instantly. It runs in the background, whenever Kafka decides. So your system works fine... until it doesn't.

Scenario: Three failures for the same key

T1: Event A fails  → Produce Key="Order-123", Value="LOCKED"
T2: Event B fails  → Produce Key="Order-123", Value="LOCKED"
T3: Event C fails  → Produce Key="Order-123", Value="LOCKED"

Local state: count = 3 ✓ (correct, you saw all three)

Hours later, Kafka compaction runs.
Three records with Key="Order-123" → collapsed into one.

T4: Pod restarts, replays lock topic to rebuild state.

Consumer sees one "LOCKED" record. Count = 1. (wrong!)

T5: Event A succeeds → Tombstone deletes the only record.

Count = 0. System thinks Order-123 is free.
But B and C are still in the retry queue.

The local in-memory state was correct. The problem only surfaces after a restart — when you replay the compacted topic and your counts are silently wrong.

Fix: use a unique ID for each lock operation.

By using a UUID as the Kafka key and putting the business key in headers, you force Kafka to keep all three lock records distinct. After restart, the consumer replays three separate records and counts correctly.

func (c *KafkaCoordinator) Acquire(ctx context.Context, msg Message) error {
    lockID := uuid.New().String()  // Unique per lock operation

    c.local.Acquire(ctx, msg)

    return c.producer.Send(ctx, c.lockTopic, &LockMessage{
        Key:   lockID,           // Kafka key = UUID (unique, never collides)
        Value: lockID,           // Lock marker
        Headers: map[string]string{
            "business-key":      msg.Key(),  // Actual key in headers
            HeaderCoordinatorID: c.instanceID,
        },
    })
}

The consuming side reads the business-key header to know which business key the lock applies to.

The trade-off you need to understand: UUID keys mean compaction can never deduplicate lock records — every record has a unique key by definition. The lock topic now grows with every lock and unlock operation. With high failure rates, this topic can grow fast, which directly impacts cold start time (Problem 5) and memory usage (covered in Part 3). Tune your retention aggressively.

Problem 5: The Rebalancing Race

Pod A crashes. Kubernetes starts Pod B. Kafka assigns Partition 0 to Pod B.

Pod B's local state is stale. It hasn't finished reading the lock topic, so it doesn't know Order-123 is locked. It processes messages out of order.

The Synchronization Barrier

A consumer must not process messages until its view of the lock state is current.

func (c *KafkaCoordinator) Synchronize(ctx context.Context) error {
    // Step 1: Ask broker for the high water mark (latest offset)
    hwm, err := c.admin.GetHighWaterMark(ctx, c.lockTopic)
    if err != nil {
        return fmt.Errorf("failed to get high water mark: %w", err)
    }

    // Step 2: Wait until our local consumer catches up
    for {
        currentOffset := c.lockConsumer.CurrentOffset()

        if currentOffset >= hwm {
            return nil  // Caught up, safe to proceed
        }

        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(100 * time.Millisecond):
            // Keep waiting
        }
    }
}

Hook this into the consumer's setup phase:

func (h *Handler) Setup(session sarama.ConsumerGroupSession) error {
    // Block until lock state is current
    if err := h.coordinator.Synchronize(session.Context()); err != nil {
        return fmt.Errorf("failed to synchronize state: %w", err)
    }
    return nil
}

Now a pod never "guesses" the lock state. It waits until it knows.

The Rebalance Loop

This barrier has teeth. Every time a pod restarts or Kafka rebalances partitions, the consumer blocks until it catches up on the lock topic. With thousands of active locks, this adds seconds to startup. With millions — especially if the lock topic has grown due to the UUID trade-off from Problem 4 — it can trigger a rebalance loop: you exceed max.poll.interval.ms waiting to sync, Kafka kicks you out, the next pod inherits the partition and does the same thing, and the partition becomes effectively frozen.

For high-churn systems, you need to tune retention on the lock topic aggressively, or move to an external state store (covered in Part 3).

What About New Locks During Sync?

New locks may arrive after the high water mark snapshot. But this doesn't cause a race — and the reason is worth understanding.

The lock is always written before the failed message gets redirected to the retry topic, and the main consumer commits its offset after the redirect. So the sequence is: lock written → message redirected → offset committed. When Pod B takes over the partition, it starts consuming from the last committed offset. Any message that needs a lock was already redirected by Pod A before the offset advanced. Pod B won't encounter that message on the main topic — it's already in the retry topic.

For messages that arrive after Pod B starts, the lock consumer is already tailing the lock topic continuously. Since the lock is written before the redirect, Pod B will see the lock before the redirected message could possibly arrive.

What About the Retry Consumer?

The retry consumer doesn't need a sync barrier. Its job is to process messages that are already in the retry topic and release locks when they succeed. It doesn't check whether a key is locked — every message in the retry topic is there precisely because the key is (or was) locked. It processes sequentially within each partition, releases on success, and re-publishes on failure. No lock lookup needed, no stale state risk.

Summary

Implementing ordered retries isn't routing messages to different topics. It's building a distributed state machine on top of Kafka — and distributed state is never consistent when you need it to be.

The pattern looks clean on a whiteboard. This is what happens when you actually build it.

Dual-Write — Use transactions or compensating actions
Reference Counting — Locks are counters, not booleans
Self-Consumption — Tag messages with origin ID, ignore your own
Compaction Collision — UUID per lock, business key in headers (but this disables compaction deduplication)
Rebalancing Race — Sync barrier before processing

The full implementation is in kafka-resilience. The code snippets here are simplified — the real version handles more edge cases.

In Part 3, we'll look at the operational cost: cold starts, memory pressure, and why you might choose a simpler solution instead.

Ordered Retries in Kafka: Why The Retry Topic Is Breaking Downstream

Myroslav Vivcharyk — Tue, 10 Mar 2026 15:10:00 +0000

I was on-call when the P1 hit. Slack lighting up, a downstream team's database rejecting writes. Constraint violations everywhere.

I pulled up the logs. The events looked fine. The schema was correct. The data made sense. Except the timestamps didn't.

10:00:01 - Event 1: ADD 100 units to SKU-123
10:00:03 - Event 2: REMOVE 50 units from SKU-123

That's the sequence that happened in the real world. Here's what the consumers saw:

10:00:01 - Event 1: ADD 100 → FAILED (API timeout) → sent to retry topic
10:00:03 - Event 2: REMOVE 50 → FAILED (CHECK constraint: quantity < 0)
10:03:01 - Event 1: ADD 100 → SUCCESS (from retry topic, 3 minutes late)

The downstream inventory table had CHECK (quantity >= 0). Because Event 2 arrived before Event 1 was retried, the system tried to remove stock that didn't exist yet. PostgreSQL said no.

The retry logic did exactly what it was designed to do — keep the pipeline moving by shoving failed messages aside. In doing so, it broke the ordering that every downstream consumer depended on.

The postmortem took two days. The fix took even more. The root cause? Your retry decisions propagate to every downstream consumer.

The Downstream Ripple Effect

When you consume from a topic and publish to another, you aren't just moving data. You're a guardian of ordering semantics for everyone downstream.

The team consuming your enriched stream might be:

Building a state machine — order status must flow CREATED → PAID → SHIPPED, never backwards
Calculating running totals — account balances or inventory counts where State(N) depends on State(N-1)
Enforcing causality — you can't update a record that hasn't been "created" yet

The worst part? You won't know you've broken ordering until production data starts drifting. Or worse — a 2 AM call about corrupted state that's been silently building for weeks.

The Two Bad Options

If your operations are idempotent and commutative — A then B gives the same result as B then A — the standard retry-topic-and-DLQ pattern works fine. Uber does it. Confluent shows it.

But for stateful systems where order matters, you're stuck choosing between two bad options:

Option 1: Stop the World. Don't commit the offset. Retry in place until it succeeds.

func (h *Handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        for {
            err := h.process(session.Context(), msg)
            if err == nil {
                break
            }
            time.Sleep(h.backoff.Next())
        }
        session.MarkMessage(msg, "")
    }
    return nil
}

Order is preserved. But one poison pill blocks the entire partition — if offset 10 fails for three hours, offsets 11 through 1,000,000 sit waiting. Your client's background heartbeat thread won't save you — max.poll.interval.ms is a separate timer, and if you exceed it, Kafka assumes your consumer is livelocked, kicks it from the group, and triggers a rebalance storm.

Option 2: Skip and Pray. Send the failed message to a retry topic. Commit. Keep going.

func (h *Handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        if err := h.process(session.Context(), msg); err != nil {
            h.sendToRetryTopic(session.Context(), msg)
        }
        session.MarkMessage(msg, "")
    }
    return nil
}

Throughput maintained. Ordering sacrificed. This is what caused my P1.

The Third Way

We need the availability of Option 2 with the ordering guarantees of Option 1.

Picture a restaurant kitchen during dinner rush. Table 5 orders appetizers, main, dessert — in that order. The steak comes out wrong and needs a redo. You don't shut down the kitchen. You don't stop serving every other table. You hold Table 5's remaining courses until the steak is ready. Tables 6, 7, and 8 keep getting their food on time.

Each table is a message key. A failed dish doesn't block the kitchen — it blocks that table's order until resolved.

When a message for key K fails, all subsequent messages for key K must wait — but messages for other keys proceed normally.

This pattern is described in a Confluent blog post. The concept is straightforward. The implementation is where things get interesting.

The Architecture

To make this work across multiple consumer instances — say, 50 pods in Kubernetes — we need a way to coordinate which keys are currently blocked. A pod that locks a key must somehow tell every other pod. And when a pod crashes mid-retry, the lock must survive.

We'll use Kafka itself as the coordination layer. On top of your original topic, you need two additional topics:

Retry Topic — messages land here either because they failed or because a predecessor with the same key failed. A separate consumer processes them with backoff. After exhausting retries, messages move to a DLQ — that's a standard pattern we won't rehash here.
Lock Topic (compacted) — the shared lock registry. Every lock and unlock is a record here.

A retry topic alone isn't enough. It handles the failed message — but the main consumer doesn't know that SKU-123 is in trouble. When the next message for SKU-123 arrives, it processes immediately, out of order. We need a way to tell every consumer instance: "this key is blocked, don't touch it." That's what the lock topic does.

How Lock State Propagates

When a message fails, the consumer writes a record to the lock topic: key="SKU-123", value="LOCKED". When the retry eventually succeeds, it writes a tombstone (null value) to signal the lock is cleared.

Every consumer instance subscribes to the lock topic and builds a local in-memory map from the stream. Because the topic is compacted, Kafka periodically deduplicates records — keeping only the latest value per key. A new instance can read from the beginning and reconstruct the full lock state. The topic is the database.

This means each instance runs two consumer loops: one for business messages (main or retry), and one that continuously tails the lock topic to keep the local map in sync.

The Lock Check

Before processing any message, the consumer checks its local map: "Is this key currently blocked?"

New message arrives: key="order-123"
│
├─ Lock map: "order-123" locked?
│   │
│   ├─ YES → Redirect to retry topic
│   │        (preserve order behind predecessor)
│   │
│   └─ NO  → Process normally

If the key is locked, the message gets redirected to the retry topic — not because it failed, but because a predecessor with the same key failed. It lines up behind that predecessor and waits its turn.

Seeing It In Action

Here's how the full flow handles the scenario from the opening:

Event 3 (different key) was never blocked. Events 1 and 2 (same key) maintained their order despite the failure. The partition kept flowing.

Where It Breaks

That diagram shows the happy path. In a distributed system with 50 pods, there are at least five ways this breaks:

1. The Dual-Write. When a message fails, you need to write a lock and publish the message to the retry topic. If you crash between the two, you've created a zombie lock — a key locked forever with no retry message that will ever release it. Do you wrap both writes in a Kafka transaction? Or is there something cheaper?

2. The Reference Counting Gap. Two messages for the same key fail. You lock the key. The first retry succeeds — do you release the lock? If you do, the third message for that key processes immediately, before the second retry. The lock isn't a boolean. It's a counter.

3. The Self-Consumption Trap. Every instance publishes lock records and subscribes to the lock topic. Kafka doesn't filter — you'll consume your own messages. Apply the lock locally and from the broadcast, and you've double-counted every lock.

4. The Compaction Collision. Three failures for the same key produce three lock records with the same Kafka key. Compaction collapses them into one. After a restart, you replay one record instead of three. Your reference count is silently wrong.

5. The Rebalancing Race. A new pod starts, gets assigned a partition, and begins processing. But it hasn't finished reading the lock topic yet. It doesn't know what's locked. It processes a message out of order.

None of these are theoretical. They all showed up during implementation.

What's Next

I've built a Go library — kafka-resilience — that handles these edge cases.

In Part 2, we'll walk through each of the five problems above and their fixes. Code included.

Part 3 is the counterargument: operational costs, failure modes, and why you might choose a simpler solution.

The goal isn't to convince you to use this — it's to give you the full picture so you can make an informed decision.

The Ghost in the Database: Why Is This Empty Table Taking 20 Seconds to Query?

Myroslav Vivcharyk — Tue, 03 Feb 2026 15:05:00 +0000

SELECT * FROM my_table;

Query execution time: ~5 seconds

Table rows: 0

This made absolutely no sense.

A few years ago, I found myself staring at a performance metric that defied logic. We had a table that was, for all intents and purposes, empty. Yet, querying it was taking longer than a coffee break.

The Setup

Here's the context: one massive MySQL database attached to an old monolith. Multiple teams outside of monolith had read-only access. The database was running on one of the beefiest RDS instances available - typically at 50-60% load.

What bothered me was that this load level never quite matched our actual traffic patterns. Something felt off.

Our monolith used the transactional outbox pattern - storing records in MySQL before publishing to RabbitMQ. The setup was straightforward: the monolith writes messages to an outbox_message table, and a separate worker service reads them in batches, publishes to RabbitMQ, then deletes them.

High throughput table. Lots of writes. Lots of deletes. Should be simple.

Then these warnings started appearing:

[WARN] Slow batch processing detected: 23.4s
[WARN] Slow batch processing detected: 31.2s
[WARN] Slow batch processing detected: 47.8s

Confirming the Symptoms

The monitoring dashboard showed the first signs. The outbox worker's latency metric - which measures the time to fetch messages from the database and publish them to RabbitMQ - had jumped to high levels. We were seeing 20+ seconds with spikes up to 50 seconds.

Distributed traces pointed to the culprit: the SELECT query that fetches message batches was taking almost all the time. The commit operation was also suspiciously slow.

Time to verify directly:

SELECT * FROM outbox_message LIMIT 1;

An empty table taking 5 seconds to query.

I checked RDS Performance Insights expecting to see obvious problems - CPU maxed out, disk I/O bottlenecks, something.

Nothing. The database was sitting at a steady 50-60% load. It looked "busy," but not "breaking." No unusual spikes. The Top Queries view didn't even show our problematic query.

Which was actually the first real clue. This wasn't a typical bottleneck.

Following the Clues

The MySQL slow query log revealed what Performance Insights had missed:

# Query_time: 18.814028  Lock_time: 0.000021 Rows_sent: 1000  Rows_examined: 1000
SELECT * FROM outbox_message LIMIT 1000 FOR UPDATE SKIP LOCKED;

18.8 seconds for a query that examined only 1000 rows. The lock time was negligible.

My first theory: blocking transactions. The worker retrieves messages in batches concurrently, so maybe they're stepping on each other.

SELECT trx_id, trx_state, trx_started, trx_query,
       dl.lock_type, dl.lock_mode, dl.LOCK_STATUS
FROM information_schema.innodb_trx
JOIN performance_schema.data_locks dl ON trx_id = ENGINE_TRANSACTION_ID
WHERE trx_query LIKE '%outbox_message%'
ORDER BY trx_started DESC;

Yes - several long-running transactions on the outbox_message table. But here's the thing: no deadlocks. All transactions had GRANTED locks, and all lock types were compatible. No conflicts.

The locks weren't the problem.

Let's look at the table itself:

SELECT table_name, 
       sys.format_bytes(data_length) AS data_size,
       sys.format_bytes(data_free) AS free_space,
       TABLE_ROWS
FROM information_schema.tables
WHERE TABLE_NAME='outbox_message';

data_size: 1.2 GiB
free_space: 1.1 GiB
TABLE_ROWS: 0

An empty table occupying 1.2 GiB on disk. Something was fundamentally wrong with this particular table.

Time to look deeper:

SHOW ENGINE INNODB STATUS;

And there it was:

History list length 21,842,680

21 million entries in the history list.

Understanding What's Actually Happening

Before we go further, let's step back and understand what this history list means.

Multi-Version Concurrency Control (MVCC)

InnoDB uses MVCC to allow concurrent reads and writes without locking. When you update or delete a row:

InnoDB doesn't immediately overwrite or remove the old version
It creates a new version of the row (or marks it as deleted)
The old version gets stored in the undo log
A reference to this old version is added to the history list

This enables readers to access old versions while writers work on new ones - no blocking required.

InnoDB's purge operations run in the background to clean up old row versions. But here's the critical constraint: the purge thread can only remove versions that are older than currently active transactions.

When you start a transaction, InnoDB creates a "snapshot" at that point in time. Even if your transaction sits idle, MySQL must keep all row versions created after your transaction started - because you might need them for consistent reads.

The Domino Effect

Here's what happened in our case:

Three transactions started. Due to a bug in another service, they never completed - no commit, no rollback. They just sat there, open for hours.

Meanwhile, the outbox worker continued its normal operation:

Write message to outbox_message (creates new row version)
Read and process message
Delete message (creates another row version marking the row as deleted)
Repeat millions of times

The purge thread's dilemma: It could see all these deleted row versions in the history list. It wanted to clean them up. But it couldn't - those three transactions were still active. MySQL had to keep every single row version created since they started.

The result: The history list grew to 21 million entries at that moment. The table physically contained gigabytes of old row versions for rows that had been processed and deleted.

The query impact: When our outbox worker ran SELECT * FROM outbox_message, InnoDB had to scan through 21 million old versions to determine which rows should be visible to this transaction.

This is why an empty table was taking 20 seconds to query. We weren't querying an empty table - we were querying through 21 million historical versions of rows that didn't exist anymore.

And here's the kicker about that 50-60% database load I mentioned earlier: it wasn't doing useful work. The database was spending enormous resources on background that was going nowhere.

Finding the Smoking Gun

Now that I understood the problem, I needed to find what was preventing the purge. Let's find those hanging transactions:

SELECT ps.id AS processlist_id,
       trx_id, 
       trx_started,
       TIMESTAMPDIFF(HOUR, trx_started, NOW()) AS hours_running,
       trx_query
FROM information_schema.innodb_trx trx
JOIN information_schema.processlist ps ON trx.trx_mysql_thread_id = ps.id
WHERE trx.trx_started < CURRENT_TIME - INTERVAL 10 MINUTE
ORDER BY trx_started;

There they are: three transactions that had been running for looong time.

These queries were coming from another service that had database access - a bug in their code was leaving transactions open, preventing MySQL from cleaning up millions of row versions for rows that had been processed and deleted.

And here was the systemic issue hiding in plain sight:

SHOW VARIABLES WHERE Variable_name IN (
    'innodb_max_purge_lag',
    'innodb_max_purge_lag_delay',
    'innodb_purge_threads' 
);

    innodb_max_purge_lag: 0
    innodb_max_purge_lag_delay: 0
    innodb_purge_threads: 4

innodb_max_purge_lag = 0 means the history list can grow indefinitely. No guardrails. No limits.

The Resolution

Now that we understood the problem, the fix was straightforward:

Immediate actions:

Kill the hanging transactions
Wait for the purge thread to catch up (took sime time to process 21 million entries)
Optimize the table to reclaim disk space

Within minutes, query performance returned to normal. And the database load? It dropped from 50-60% down to 10%. Same traffic — but now the database wasn't wasting resources on futile purge operations.

Long-term fixes:

Configure proper purge limits:

    SET GLOBAL innodb_max_purge_lag = XXX;

This prevents the history list from growing beyond XXX entries. If it reaches this limit, InnoDB adds small delays to DML operations to let the purge thread catch up. A small performance may be better than a complete breakdown.

Fix transaction timeout handling across all services with database access. No transaction should run for hours/days.
Add monitoring and alerts for history list length. This metric should be on every write-heavy database dashboard.

    SELECT count FROM information_schema.INNODB_METRICS
    WHERE name = 'trx_rseg_history_len';

Wrapping Up

You can't control everything in a large systems. Bugs will happen. But what you can control:

1. Monitoring Matters

Add HLL metric to your database dashboard:

SELECT count FROM information_schema.INNODB_METRICS 
WHERE name = 'trx_rseg_history_len';

Or to your Performance insights dashboard.

2. Set Proper Guardrails

Default MySQL configurations assume perfect transaction hygiene. In complex systems with multiple services, you may need limits.

Don't let innodb_max_purge_lag stay at 0 in write-heavy databases with high delete volumes. Set it to something reasonable and accept the small performance trade-off for the reliability gain.

3. Understand the Internals

Knowing how MVCC and the purge thread work was important to diagnosing this issue. The history list concept isn't obscure trivia - it's fundamental to how InnoDB handles concurrency.

Every DELETE doesn't immediately remove data; it just marks a new version. Every UPDATE creates a new row version. All of this goes into the history list until the purge thread can safely remove it.

4. Discipline Matters

The service that caused this only had read-only access. But even a read-only service can bring down your database if it doesn't properly manage transactions.

Every service needs proper timeout handling and error recovery that ensures transactions complete. No exceptions.

Resources

Go Graceful Shutdown: Beyond ctx.Done()

Myroslav Vivcharyk — Tue, 13 Jan 2026 15:15:00 +0000

As Go engineers, we've all internalized the context pattern. When an application shuts down, we propagate the cancellation, our select blocks catch <-ctx.Done(), and we return. Clean. Simple. Idiomatic.

It usually looks like this:

for {
    select {
    case <-ctx.Done():
        return // Clean exit, right?
    default:
        // Do work
    }
}

For many applications, this is "good enough." If you're running a stateless process that can safely restart mid-operation, Kubernetes will just spin up a new POD and life goes on. A few "context canceled" errors in your logs? Who cares.

But what about processing a payment when SIGTERM arrives? Or acknowledging a message you've half-processed?

When "good enough" isn't enough anymore?

Not every application can afford to drop everything and exit. Consider what happens when you're halfway through processing a payment and SIGTERM arrives. Or when you've pulled a message from a queue but haven't acknowledged it yet.

Here are few scenarios where naive context cancellation creates real problems:

Message queues with acknowledgments — SQS, RabbitMQ, Kafka — any system where you must explicitly confirm message processing. Unacknowledged messages get reprocessed, causing duplicates or delays.

Multi-step transactions — Payment captured but not recorded. Inventory reserved but not confirmed. Customer charged but order not created.

Non-idempotent operations — Notifications already sent. Resources already created. Webhooks already fired. You can't just "try again."

Long-running batch jobs — Hours of computation that can't resume from the middle. Data migrations left in inconsistent states.

In all these cases, the cost of interruption exceeds the cost of waiting a few extra seconds for clean completion.

If your code lives in any of these worlds, keep reading.

The Problem: One Signal, Two Jobs

Let's look at a typical message processing pipeline - I'll use SQS as an example since it clearly demonstrates the issue, but the pattern applies to any acknowledge-based system:

Here's the typical implementation:

func ConsumeMessages(ctx context.Context, queueURL string) error {
    messages := make(chan Message, 100) // Buffered channel

    go pollMessages(ctx, queueURL, messages)

    for {
        select {
        case <-ctx.Done():
            // Problem: returns immediately!
            // Buffered messages abandoned
            // In-flight processing interrupted
            return ctx.Err()

        case msg := <-messages:
            processMessage(ctx, msg)    // ← SIGTERM during this
            deleteFromSQS(ctx, msg)     // ← Never executes
        }
    }
}

When SIGTERM arrives and ctx.Done() fires, here's what may happen:

func processMessage(ctx context.Context, msg Message) error {
    // Step 1: Call external API
    if err := callSomeAPI(ctx, msg.Data); err != nil {
        return err  // ctx.Done() could interrupt here
    }

    // Step 2: Database save interrupted by ctx.Done()
    if err := saveToDatabase(ctx, msg.Data); err != nil {
        return err  // or here
    }

    return nil
}

Since we propagate the context to processMessage(), the context cancellation flows through your function calls. The external API call might complete, but the database save fails with context.Canceled. Now you have partial state — the API side effect happened, but nothing was saved. And since acknowledgeMessage() never runs, the message gets reprocessed, potentially duplicating the API call.

Additionally, any messages already fetched from the queue but sitting in a buffer waiting to be processed get abandoned when the loop exits — they'll timeout and reappear for reprocessing.

The root issue: We have one signal (ctx.Done()) trying to do two completely different jobs.

Stop accepting new work
Stop working immediately

We need to separate these concerns.

Mental Model: The Restaurant Kitchen

Imagine your system like a restaurant kitchen during closing time:

  Orders In  →  Prep Station  →  Cooking  →  Serve to Customer
   (fetch)       (buffer)       (process)     (complete)

Naive shutdown (ctx.Done()):

The manager shouts "We're closed!" and the chef drops the pan mid-stir and walks out.

Orders on the counter? Tossed.
Food on the stove? Abandoned half-cooked.
Customers waiting for meals? Kicked out hungry.

Good shutdown (graceful):

Lock the front door — Stop taking new orders
Finish what's cooking — Complete dishes on the stove
Serve completed meals — Deliver to waiting customers
Then turn off the lights and go home

Of course, you can't wait forever. If it's been 30 minutes and someone's still cooking, you'll eventually have to call it. But you give the kitchen a reasonable chance to finish.

This is exactly our strategy: control the shutdown signal so work can flow to completion.

The Two-Phase Shutdown

The solution is elegantly simple: use two signals instead of one.

Signal 1: "Please stop gracefully" — Stop accepting new work

Signal 2: "Emergency brake" — Force shutdown if graceful fails (via ctx.Done())

Here's the conceptual flow:

// The core idea
select {
case <-ctx.Done():
    // Emergency: parent context canceled, stop everything now
    return ctx.Err()

case <-stopSignalCh:
    // Graceful shutdown in three phases:

    // Phase 1: Stop accepting new work
    cancelPoller()

    // Phase 2: Wait for poller to finish and close the work channel
    <-pollerDone

    // Phase 3: Processor drains remaining messages from channel
    <-processorDone

    return nil
}

We're using the simple channel close as a coordination mechanism. When the poller stops, it closes the message channel. The processor, ranging over that channel, naturally drains any remaining messages and then exits. No complex synchronization needed — just Go's built-in channel semantics.

Implementation: A Real Example

Let me show you how this looks in practice. I'll use a real SQS consumer implementation that I built - you can see the full code on GitHub, but I'll walk through the key parts here.

The Data Flow

Before diving into code, let's visualize what we're building:

Single Message Flow:
┌──────┐    ┌─────────┐    ┌─────────┐    ┌────────┐
│ Poll │───▶│Transform│───▶│ Process │───▶│ Delete │
│ SQS  │    │ Message │    │Handler  │    │  (ACK) │
└──────┘    └─────────┘    └─────────┘    └────────┘

Graceful Shutdown Flow:

stopSignalCh receives signal
Poller stops fetching → closes channel
Processor sees closed channel → drains remaining messages
All messages acknowledged → Exit

Core Structure

We need to track both the shutdown request and completion:

type Consumer[T any] struct {
    poller      Poller
    processor   Processor[T]

    stopSignalCh chan struct{}  // Trigger graceful shutdown
    stoppedCh    chan struct{}  // Signal completion

    // ... config, logging, etc
}

The Main Loop: Control Tower

Here's the heart of the system:

func (c *SQSConsumer[T]) Consume(
    ctx context.Context,
    queueURL string,
    handler Handler[T],
) error {
    // Buffer size balances throughput against shutdown time
    bufferSize := c.cfg.ProcessorWorkerPoolSize * 3
    msgs := make(chan sqstypes.Message, bufferSize)

    // Separate contexts let us control shutdown independently
    processCtx, cancelProcess := context.WithCancel(ctx)
    pollerCtx, cancelPoller := context.WithCancel(ctx)

    // Cleanup always happens, whether we exit gracefully or via context cancellation
    defer func() {
        cancelPoller()
        cancelProcess()
        c.isRunning = false
        c.metrics.RecordShutdown(ctx)
    }()

    c.isRunning = true

    pollerErrCh := make(chan error, 1)
    processErrCh := make(chan error, 1)

    // Start both components
    go func() { pollerErrCh <- c.poller.Poll(pollerCtx, msgs) }()
    go func() { processErrCh <- c.processor.Process(processCtx, msgs, handler) }()

    // The control tower
    select {
    case <-ctx.Done():
        // Emergency brake—parent context canceled (e.g., timeout expired)
        return ctx.Err()

    case <-c.stopSignalCh:
        // Graceful shutdown sequence

        // PHASE 1: Stop accepting new work
        cancelPoller()

        // PHASE 2: Wait for poller to close the channel
        // CRITICAL: The poller MUST close 'msgs' when it exits
        if err := <-pollerErrCh; err != nil {
            return err
        }

        // PHASE 3: Processor drains remaining messages
        // It exits naturally when the channel closes
        if err := <-processErrCh; err != nil {
            return err
        }

        // Signal that shutdown is complete
            // This unblocks Close() which is waiting on this channel
        close(c.stoppedCh)

      return nil
    }
}

The key insight: we cancel only the poller's context (closing the door). The processor keeps running until the channel is empty and closed. The processor doesn't know or care that we're shutting down — it just keeps processing until there's no more work.

The Poller: Locking the Front Door

Remember our restaurant analogy? The poller is the host who locks the door. Its job is simple but critical — it must close the channel when it exits:

func (p *Poller) Poll(ctx context.Context, ch chan<- Message) error {
    defer close(ch)  // This one line coordinates everything

    for {
        if ctx.Err() != nil {
              // Context canceled—time to stop
            return nil
        }

        // Poll for messages
        messages := p.fetchMessages(ctx, queueURL)

        for _, msg := range messages {
            select {
            case ch <- msg:
                // Message sent to processor
            case <-ctx.Done():
                // Context canceled while sending
                return nil
            }
        }
    }
}

That defer close(ch) does the heavy lifting. When the poller's context is canceled:

The function returns
The defer closes the channel
The processor's range loop exits naturally

No explicit coordination needed.

The Processor: Finishing What's on the Stove

The processor simply ranges over the channel until it closes:

func (p *Processor[T]) Process(ctx context.Context, msgs <-chan Message, handler Handler[T]) error {
    var wg sync.WaitGroup

    // Start worker pool
    for i := 0; i < p.workerPoolSize; i++ {
        wg.Add(1)

        go func() {
            defer wg.Done()

            // Process messages until channel closes
            for msg := range msgs {
                select {
                case <-ctx.Done():
                        // Emergency exit if timeout expires
                    return
                default:
                    p.processMessage(ctx, msg, handler)
                }
            }
        }()
    }

    // Wait for all workers to finish
    wg.Wait()

    return nil
}

The processor keeps the ctx.Done() check in case we hit the timeout and need to force-exit, but during normal graceful shutdown, it simply processes until msgs closes.

The Close Method: Initiating Shutdown

This is the public API your main.go calls when it catches SIGTERM:

func (c *Consumer[T]) Close() error {
    c.mu.Lock()

    if !c.isRunning || c.isClosing {
        c.mu.Unlock()
        return nil
    }

    c.isClosing = true
    c.stopSignalCh <- struct{}{}  // Trigger graceful shutdown
    c.mu.Unlock()

    // Wait for completion, but not forever
    timeout := time.Duration(c.cfg.GracefulShutdownTimeout) * time.Second

    select {
    case <-c.stoppedCh:
        // Clean shutdown completed
        return nil

    case <-time.After(timeout):
        // Timeout expired—shutdown incomplete
        return fmt.Errorf("shutdown timeout after %v", timeout)
    }
}

The timeout is crucial. In the real world, you can't wait indefinitely. Kubernetes will eventually send SIGKILL if your POD doesn't terminate. The timeout gives you a chance to clean up, but ensures you don't hang forever.

Choosing a Timeout Value

How do you pick a good timeout?

shutdownTime = (bufferSize / workerPoolSize) × avgProcessingTime × safetyFactor

Example:

Buffer: 30 messages
Workers: 10
Processing time: 100ms per message
Safety factor: 3x

shutdownTime = (30 / 10) × 100ms × 3 = 900ms

Set timeout to 2-3x this: GracefulShutdownTimeout = 3 seconds

What Happens When The Timeout Expires?

If the timeout expires, Close() returns an error, but the consumer is still running. At this point, you have a few options:

Let it go — Kubernetes will SIGKILL the POD shortly.
Force cancel — Cancel the parent context to trigger the emergency brake.

In production code, this looks like:

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    consumer := NewConsumer(cfg, poller, processor)

    // Handle shutdown signals
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)

    go func() {
        <-sigCh
        log.Info("Shutdown signal received")

        if err := consumer.Close(); err != nil {
                // Decision point: what now?          
            log.Error("some messages may be reprocessed", "error", err)
            cancel()  // Emergency brake: force everything to stop
        }
    }()

    // This blocks until shutdown completes
    if err := consumer.Consume(ctx, handler); err != nil {
        log.Error("Consumer error", "error", err)
    }
}

Key Takeaways

The problem: ctx.Done() can't distinguish "stop accepting work" from "finish current work." This may leave in-flight work in inconsistent states.

The solution:

Use separate contexts for different components
Use a dedicated signal channel for graceful shutdown
Let channel closure coordinate the drain naturally
Always include a timeout — you can't wait forever

When to use this pattern:

Message queue consumers (SQS, RabbitMQ, Kafka)
Any system with explicit acknowledgments
Multi-step operations that must complete atomically
Long-running batch jobs
Non-idempotent external API calls

When you don't need it:

Stateless operations
Idempotent operations that can safely restart
Pure computational tasks with no side effects

What's Next?

If you're building systems with in-flight work, consider what your "units of work" look like. What state do they maintain? What happens if they're interrupted? How long does completion typically take?

The answers to these questions will guide your shutdown strategy. Sometimes immediate exit is fine. Sometimes you need another approach. The key is making a conscious choice rather than defaulting to ctx.Done() everywhere.

The approach I've shown here is specific to message consumers, but the principle applies broadly.

For the complete implementation check out the full code on GitHub.

And if you've implemented graceful shutdown in a different way, or have real-life stories about what happens when you don't, I'd love to discuss them.

Kafka Consumer Health Checks: Why Progress Beats Lag

Myroslav Vivcharyk — Tue, 16 Dec 2025 14:02:20 +0000

Many of you have been there - it’s 3 AM, your phone buzzes, and you’re staring at an alert: “Kafka consumer lag exceeded threshold.”

You stumble to your laptop, check the metrics, and… everything looks fine. The messages are processing. The lag was just a temporary spike from a slow downstream service. You silence the alert and go back to bed, mentally adding another item to your "fix someday" list.

Sound familiar?

Here's what's actually happening: lag tells you how far behind you are, but not whether you're making progress. A consumer can sit at 1000 messages lag for 10 minutes because it's stuck, or because it's processing at exactly the rate messages arrive. From lag alone, you can't tell the difference.

The real question isn't "how much lag?" — it's "are we making progress?"

That’s the problem I found myself wrestling with a while back. After dealing with false positive alerts and delayed detection of real issues, I discovered a simple but brilliant solution from PagerDuty’s engineering team. Rather than retelling their story, I want to show you how to implement this approach in Go, with some insights I’ve picked up along the way.

💡 Want to dive straight into the code? Check out the complete source code here.

Why Traditional Health Checks Fall Short

Let’s talk about common patterns and why they don’t quite work:

The “Always Healthy” Approach

func healthHandler(w http.ResponseWriter, r *http.Request) {
    w.WriteHeader(http.StatusOK)
    w.Write([]byte("OK")) // "I'm alive!" (Are you though?)
}

This tells you nothing. Your consumer could be completely frozen, and Kubernetes (any orchestrator) would happily keep it alive.

The “Ping the Broker” Approach

This is better than nothing — at least you’re verifying connectivity. But connectivity doesn’t mean your consumer is processing messages. Your network might be fine while your consumer group is stuck in an infinite rebalance loop.

The “Lag Threshold” Trap

This is where most teams land. You’re probably already tracking consumer lag through Prometheus or similar tools. You’ve set up alerts when lag exceeds some threshold — maybe 100 messages, maybe 1000, maybe a million.

But here’s the problem: what should that threshold be?

Set it too low → You wake up at 3 AM because a downstream API responded slowly for 30 seconds. The consumer is fine, just briefly overwhelmed. False alarm.

Set it too high → You miss genuine issues until they’ve already cascaded into customer-visible problems. By the time your alert fires, you’re in damage control mode.

The granularity problem: A single stuck POD in your deployment of 50 can go unnoticed if you’re monitoring average lag. That POD quietly fails while the other 49 keep working, masking the issue in your aggregate metrics.

The fundamental tension: fast detection means false positives, reliable alerts mean delayed detection.

You can build sophisticated heuristics or even ML models to detect anomalies, but all of that adds complexity and still detects problems with noticeable delay.

The Key Insight: Progress vs. Position

What we really need is a way to answer a simple question: Is this specific consumer instance making progress?

Instead of measuring lag, we measure progress. Here's the approach:

The Heartbeat: Track the timestamp of the last processed message for each partition. If we're processing messages, we're healthy.

The Verification: If enough time passes without new messages (say, X seconds), we don't panic yet. We query the Kafka broker for the latest offset. Now we can make a decision:

Consumer Offset < Broker Offset: ❌ UNHEALTHY (there are messages available, but we're not processing them — we're stuck)
Consumer Offset ≥ Broker Offset: ✅ HEALTHY (we're caught up, just waiting for more work — we're idle)

This elegantly distinguishes between a stuck consumer and a consumer that's simply idle.

How It Works: Three Scenarios

Let me show you exactly what happens in each case.

Scenario 1: Active Processing (Healthy)

When messages are flowing and your consumer is processing them, health checks are instant:

No broker queries needed — we've seen recent activity.

Scenario 2: Stuck Consumer (Unhealthy)

The consumer freezes, but messages keep arriving:

The broker query reveals there's work to do, but we're not doing it.

Scenario 3: Idle Consumer (Healthy)

The consumer is caught up, just waiting for new messages:

This is the key distinction — same timeout, different outcome based on broker state.

The beauty of this approach is its simplicity — we're not doing complex analysis or ML. We're just asking two questions: "Have I seen new work? If not, is there new work available?"

Implementation

I’ve packaged this logic into kafka-pulse-go, a lightweight library that works with most popular Kafka clients. The core logic is decoupled from specific client implementations, with adapters provided for Sarama, segmentio/kafka-go, and Confluent’s client.

Let me walk you through how to use it.

Setting Up the Monitor

The setup is straightforward. Here's what you need with Sarama:

import (
    "github.com/IBM/sarama"
    adapter "github.com/vmyroslav/kafka-pulse-go/adapter/sarama"
    "github.com/vmyroslav/kafka-pulse-go/pulse"
)

func main() {
    // Your existing Sarama setup
    config := sarama.NewConfig()
    client, err := sarama.NewClient(brokers, config)
    if err != nil {
        log.Fatal(err)
    }

    // Create the health checker adapter
    brokerClient := adapter.NewClientAdapter(client)

    // Configure the monitor
    monitorConfig := pulse.Config{
        Logger:       logger,
        StuckTimeout: 30 * time.Second,
    }

    monitor, err := pulse.NewHealthChecker(monitorConfig, brokerClient)
    if err != nil {
        log.Fatal(err)
    }

    // Pass monitor to your consumer...
}

What's StuckTimeout? This defines how long we'll wait without seeing new messages before we query the broker. Set this based on your expected message frequency:

High-throughput topics: 10-30 seconds works well
Medium-volume topics: 1-2 minutes is reasonable
Low-volume topics: You might need 5-10 minutes

The key is balancing detection speed with broker query overhead. Too short and you’ll query the broker constantly during idle periods. Too long and you’ll delay detecting real issues.

Integrating with Your Consumer

The integration is minimal — just call Track() for each message you process:

func (h *consumerGroupHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession, 
    claim sarama.ConsumerGroupClaim,
) error {
    for message := range claim.Messages() {
        // Process your business logic
        if err := processMessage(message); err != nil {
            // Handle error...
        }

        // Tell the health monitor we're making progress
        wrappedMessage := adapter.NewMessage(message)
        h.monitor.Track(session.Context(), wrappedMessage)

        session.MarkMessage(message, "")
    }
    return nil
}

That’s it. The monitor now knows when you’ve processed a message and can track progress per partition.

Exposing the Health Check

Wire this into your HTTP server for health checks:

func (hs *HealthServer) healthHandler(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
    defer cancel()

    isHealthy, err := hs.monitor.Healthy(ctx)

    if err != nil {
        // Handle broker connection errors
        w.WriteHeader(http.StatusServiceUnavailable)
        json.NewEncoder(w).Encode(map[string]string{
            "status": "unhealthy",
            "error":  err.Error(),
        })
        return
    }

    if isHealthy {
        w.WriteHeader(http.StatusOK)
        json.NewEncoder(w).Encode(map[string]string{"status": "healthy"})
    } else {
        w.WriteHeader(http.StatusServiceUnavailable)
        json.NewEncoder(w).Encode(map[string]string{
            "status": "unhealthy",
            "reason": "consumer stuck behind available messages",
        })
    }
}

Now your orchestrator (Kubernetes, ECS, whatever) can hit this endpoint and know the real health of your consumer.

The Critical Detail: Handling Rebalances

Here’s where custom implementations may fail. In Kafka, partition ownership is fluid. If you add a new POD or an old one crashes, Kafka triggers a rebalance. Your consumer instance might lose ownership of Partition 1 and gain an ownership of Partition 2.

The Danger of the "Zombie Partition"

Without proper cleanup, here's what happens:

Rebalance Triggered: You added a new POD to consumer group, or an existing one crashes. Kafka initiates a consumer group rebalance to redistribute partition ownership.
Your consumer loses ownership of Partition 1 and stops reading from it (correctly - you no longer own it)
The health checker still holds a reference to Partition 1 in its memory
Time passes without new messages for Partition 1 in your tracker (obviously — you're not reading it anymore)
The health checker eventually queries the broker and sees the offset for Partition 1 is moving (thanks to whoever owns it now)
But your local tracked offset for Partition 1 is frozen at the last message you processed before losing ownership
Result: The health checker flags your pod as UNHEALTHY for a partition it doesn't even own anymore

This is a "zombie partition" problem — your health checker is tracking ghosts.

The Fix

You must manage the partition lifecycle. Call Release() when a session ends to purge that partition from the monitor’s memory:

func (h *consumerGroupHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession, 
    claim sarama.ConsumerGroupClaim,
) error {
    ctx := session.Context()

    for {
        select {
        case <-ctx.Done():
            // CRITICAL: Clean up partition tracking during rebalance
            h.monitor.Release(ctx, claim.Topic(), claim.Partition())
            return nil

        case message := <-claim.Messages():
            processMessage(message)

            wrappedMessage := adapter.NewMessage(message)
            h.monitor.Track(ctx, wrappedMessage)

            session.MarkMessage(message, "")
        }
    }
}

The ctx.Done() signal from Sarama indicates that the session is ending — either due to a rebalance, shutdown, or error. This is your cue to clean up tracking state for this partition.

Why this matters: In a production system with frequent deployments, rebalances happen constantly. Without proper cleanup, you’ll see cascading false alerts as PODs report unhealthy for partitions they’ve lost ownership of.

Under the Hood

The library’s design is intentionally simple. Here are the key pieces:

State Tracking

We maintain a map of topic-partitions to their last-seen offset and timestamp:

type tracker struct {
    topicPartitionOffsets map[string]map[int32]OffsetTimestamp
    mu                    sync.RWMutex
}

type OffsetTimestamp struct {
    Timestamp time.Time
    Offset    int64
}

Why a map? Because each partition is independent. Partition 0 being stuck doesn’t mean Partition 1 is stuck. We need per-partition granularity to avoid false positives.

The Health Check Logic

The verification follows the two-step process:

func (h *HealthChecker) Healthy(ctx context.Context) (bool, error) {
    currentState := h.tracker.currentOffsets()
    now := h.clock.Now()

    for topic := range currentState {
        for partition := range currentState[topic] {
            offsetTimestamp := currentState[topic][partition]

            // Step 1: Has it been too long since we saw a new message?
            if now.Sub(offsetTimestamp.Timestamp) > h.stuckTimeout {
                // Step 2: Query broker for latest offset
                latestOffset, err := h.client.GetLatestOffset(ctx, topic, partition)
                if err != nil {
                    if h.ignoreBrokerErrors {
                        continue // Don't fail on transient broker issues
                    }
                    return false, fmt.Errorf("failed to get latest offset: %w", err)
                }

                // Are we stuck or just idle?
                if offsetTimestamp.Offset < latestOffset {
                    // Stuck! There are messages we haven't processed
                    return false, nil
                }
                // We're caught up, just idle - that's fine
            }
        }
    }

    return true, nil
}

The key insight is in that comparison: offsetTimestamp.Offset < latestOffset.

Practical Considerations

What About Poison Pill Messages?

A common pitfall in health checks is confusing activity with progress. If a consumer is stuck in an infinite retry loop on a single malformed message, it’s technically “active” — burning CPU and making network calls — but it’s not “healthy” because the stream is blocked.

This approach handles it correctly. Here’s why:

The tracker only updates the timestamp when the offset actually moves forward:

// Only update timestamp if offset has changed
if existing.Offset != m.Offset() || existing.Timestamp.IsZero() {
    t.topicPartitionOffsets[m.Topic()][m.Partition()] = OffsetTimestamp{
        Offset:    m.Offset(),
        Timestamp: t.clock.Now(),
    }
}

If your consumer spends 10 minutes retrying the same message at offset 100, the offset never advances. The timestamp stays frozen at the last successful message. Eventually the timeout expires, and because the broker has newer messages (101, 102, 103…), the health check correctly reports Unhealthy.

This forces visibility on “zombie loops” that may otherwise go unnoticed.

Broker Outages: Choosing Your Failure Mode

Since our verification step requires querying the broker, you need to decide what happens when that call fails.

The library exposes an IgnoreBrokerErrors flag that controls this behavior:

Option A: Fail Closed (default)

config := pulse.Config{
    StuckTimeout:       30 * time.Second,
    IgnoreBrokerErrors: false, // default - fail on broker errors
}

Upside: You never report “Healthy” when you’re actually blind to the real state
Risk: A temporary Kafka blip can cause Kubernetes to restart your entire consumer fleet simultaneously (a “restart storm”). Since restarting won’t fix the broker, this just adds chaos to an ongoing outage

Option B: Fail Open

config := pulse.Config{
    StuckTimeout:       30 * time.Second,
    IgnoreBrokerErrors: true, // ignore transient broker errors
}

Upside: Your PODs stay running, warm, and ready to resume work the moment the broker recovers
Risk: You might miss a legitimate network partition affecting only your specific POD

My recommendation: Start with fail-closed (the default). It's safer — you never report healthy when you're blind to the actual state. If you experience restart storms during broker incidents, switch to fail-open and add separate monitoring for broker connectivity at the infrastructure level.

Does This Replace Consumer Lag Monitoring?

No! Consumer lag metrics are still valuable for:

Capacity planning: Understanding if you need to scale your consumer group
Performance trending: Tracking how your processing speed changes over time
Business SLAs: If your requirement is “process messages within 5 minutes,” lag is exactly what you need to measure

This health check complements lag monitoring — it doesn't replace it. Lag tells you if you need more consumers. Health checks tell you if your existing consumers are working.

Wrapping Up

The core insight is simple: health checks should measure progress.

By distinguishing between “stuck with work to do” and “idle but caught up,” we eliminate false positives without missing real issues. No more 3 AM alerts for temporary slowdowns. No more delayed detection of genuinely stuck consumers.

The implementation is straightforward:

Track offset progression for each partition
When idle too long, verify against broker state
Handle rebalances cleanly with proper partition lifecycle management
Choose your failure mode for broker errors

Whether you use kafka-pulse-go or implement this pattern yourself, it’s universal across languages and Kafka clients. The principle remains the same.

Try It Out

The repo includes a complete working example. It's the fastest way to see how this works in practice.

API testing with simulation Vol.2

Myroslav Vivcharyk — Thu, 06 Mar 2025 17:32:28 +0000

In Part 1, we introduced Hoverfly as a tool for API simulation and explored its basic setup. We saw how it can capture API interactions and replay them during tests, making our test suite more reliable.

However, real-world applications present some real-world challenges that go beyond those basic examples. Let's be honest — if the simple scenarios from first part were sufficient for your needs, you probably don't need this series.

This part explores solutions to common challenges. We'll focus on specific scenarios that frequently appear in applications, and I believe you can adapt these examples for your own needs far beyond what we'll cover here.

Challenge One: Dynamic Data

Let's look at a test scenario:

{
    name: "create activity",
    testFunc: func(t *testing.T, client *ClientWithResponses) {
        var (
            ctx     = context.Background()
            id      = int32(1)
            title   = randomTitleForResource("activity")
            dueDate = time.Now()
        )

        resp, err := client.PostApiV1ActivitiesWithApplicationJSONV10BodyWithResponse(
            ctx,
            PostApiV1ActivitiesApplicationJSONV10RequestBody{
                Id:        tests.ToPtr(id),
                Title:     tests.ToPtr(title),
                DueDate:   tests.ToPtr(dueDate),
                Completed: tests.ToPtr(false),
            },
        )
        require.NoError(t, err)
        require.NotNil(t, resp, "expected resp, got nil")
        require.NotNil(t, resp.ApplicationjsonV10200, "expected resp body, got nil")

        assert.Equal(t, http.StatusOK, resp.StatusCode())
        assert.Equal(t, id, *resp.ApplicationjsonV10200.Id)
        assert.Equal(t, title, *resp.ApplicationjsonV10200.Title)
        assert.Equal(
            t,
            dueDate.UTC().Format(time.RFC3339),
            resp.ApplicationjsonV10200.DueDate.UTC().Format(time.RFC3339),
        )
        assert.Equal(t, false, *resp.ApplicationjsonV10200.Completed)
    },
}

With a helper function to generate random titles:

// Generate a random title to test the possibility of random data in Hoverfly
func randomTitleForResource(resource string) string {
    return fmt.Sprintf("%s-%s-title-%s", fakePrefix, resource, tests.RandomString(5))
}

A couple of points that may catch our attention:

Random data generation: The test creates a unique random title for each test run
Current timestamps: The test uses the current time as the dueDate

These scenarios are quite common in testing.

But when we run these tests with simulations, they'll fail. Why? Because our captured simulation contains specific values, but our test generates new ones each time. The simulation can't match the new requests to the captured responses.

Similarly, look at this test with UUIDs:

{
    name: "create activity with UUID in title",
    testFunc: func(t *testing.T, client *ClientWithResponses) {
        ctx := context.Background()

        // Create multiple activities with different UUIDs
        for i := 0; i < 3; i++ {
            id := int32(1 + i)
            title := uuid.New().String()

            resp, err := client.PostApiV1ActivitiesWithApplicationJSONV10BodyWithResponse(
                ctx,
                PostApiV1ActivitiesApplicationJSONV10RequestBody{
                    Id:        tests.ToPtr(id),
                    Title:     tests.ToPtr(title),
                    DueDate:   tests.ToPtr(time.Now()),
                    Completed: tests.ToPtr(false),
                },
            )
            require.NoError(t, err)
            require.NotNil(t, resp)
            require.NotNil(t, resp.ApplicationjsonV10200)
            assert.Equal(t, http.StatusOK, resp.StatusCode())
            assert.NotNil(t, resp.ApplicationjsonV10200)
            assert.Equal(t, title, *resp.ApplicationjsonV10200.Title)
        }
    },
}

Each iteration generates a new UUID, creating a unique request that won't match our captured simulation.

Understanding Hoverfly Simulations

Let's peek inside the simulation file. Hoverfly works by storing pairs of requests and responses. When it receives an incoming request, it compares it against stored requests to find a match, then returns the corresponding response. Here's what a captured request-response pair looks like:

{
  "request": {
    "path": [
      {
        "matcher": "exact",
        "value": "/api/v1/Activities/1"
      }
    ],
    "method": [
      {
        "matcher": "exact",
        "value": "GET"
      }
    ],
    "destination": [
      {
        "matcher": "exact",
        "value": "fakerestapi.azurewebsites.net"
      }
    ],
    "scheme": [
      {
        "matcher": "exact",
        "value": "https"
      }
    ],
    "body": [
      {
        "matcher": "exact",
        "value": ""
      }
    ]
  },
  "response": {
    "status": 200,
    "body": "{\"id\":1,\"title\":\"Activity 1\",\"dueDate\":\"2025-03-04T23:24:16.4781493+00:00\",\"completed\":false}",
    "encodedBody": false,
    "headers": {
      "Api-Supported-Versions": [
        "1.0"
      ],
      "Content-Type": [
        "application/json; charset=utf-8; v=1.0"
      ],
      "Date": [
        "Tue, 04 Mar 2025 22:24:15 GMT"
      ],
      "Hoverfly": [
        "Was-Here"
      ],
      "Server": [
        "Kestrel"
      ]
    },
    "templated": false
  }
}

Request Matchers

When Hoverfly captures a request, it creates a Request Matcher for each field in the request. A Request Matcher consists of:

The request field name (path, method, destination, etc.)
The type of match to use for comparison
The request field value

By default, Hoverfly sets the type of match to exact for each field, which means the incoming request must exactly match the stored request for each field. This is why random data in our tests breaks the simulation—the new random values won't exactly match the captured ones.

Hoverfly supports multiple matcher types including exact, glob, regex, and more. You can find details in Hoverfly's official documentation.

Response Templating

The templated field in the response determines whether Hoverfly should use templating. When enabled, you can inject data from the request into the response using templating syntax.

For example, this templated response includes the URL path from the original request:

"response": {
  "status": 200,
  "body": "You requested: {{ Request.Path.[0].Value }}",
  "encodedBody": false,
  "headers": {
    "Content-Type": ["text/plain"]
  },
  "templated": true
}

It's worth noting that Hoverfly's templating is far more powerful than just simple value injection. You can:

Use conditional logic with {{if}}, {{else}}
Extract data using JSONPath or XPath expressions
Perform string manipulations (substring, lowercase, etc.)
Generate random data, etc.

When Hoverfly receives a request, it will extract the values from the request body and inject them into the response, creating a consistent experience even with random data.

This is exactly what we need to solve our problems. By combining request matchers with templated responses, we can create simulations that work with any random values our tests generate.

For our dynamic data problem, we'll use these features to:

Change the matcher types for fields with dynamic data
Enable response templating to inject request values into responses
Apply these changes automatically with our simulation processor

Creating a Simulation Processor

Now that we understand how Hoverfly processes requests and responses, let's build a solution for our problems. In my code, I'll use the internal implementations from the Hoverfly package since it's open-source. However, remember this is just a JSON file, so if you use any language other than Go, you can easily implement the same approach.

The Goal

Our processor needs to achieve several objectives:

Identify dynamic data patterns in both requests and responses
Replace exact matchers with matchers for fields containing UUIDs, timestamps, prefixed random strings, etc.
Enable templating in responses to use values from the request
Support static responses for specific endpoints that need predetermined behavior

We'll use a declarative configuration file to define our processing rules, for example:

version: "1.0"
settings:
  debug: false

patterns:
  # All UUIDs will be replaced with the default regex pattern
  - type: "uuid"

  # Dates will be replaced
  - type: "datetime"
    formats:
      - "2006-01-02T15:04:05Z07:00"
      - "2006-01-02"

  # Prefix matching
  - type: "prefix"
    pattern: "fake-api-test-"
    length: 5

endpoints:
  # Replace the response of the endpoint with the static_response
  - method: "GET"
    path: "/api/v1/Activities/30"
    status: 200
    static_response: |
      {
        "id": 77,
        "title": "John Doe",
        "dueDate": "2025-02-19T17:12:52.127Z",
        "completed": false
      }

With this configuration, we aim to automatically transform our simulation files to handle dynamic data without manually editing them.

I don't want to go deep into the implementation details in this article (if you're interested, you can find everything in the code), so I'll just show you the main concepts.
Here's the entry point of the processor implementation:

func (p *PostProcessor) Process(simulation *v2.SimulationViewV5) error {
    if simulation == nil {
        return fmt.Errorf("nil simulation provided")
    }

    for i := range simulation.RequestResponsePairs {
        pair := &simulation.RequestResponsePairs[i]

        // 1. Check if this is a static endpoint rule
        if p.endpointProcessor != nil {
            if rule := p.endpointProcessor.FindMatchingRule(pair); rule != nil {
                if err := p.endpointProcessor.ApplyRule(pair, rule); err != nil {
                    return fmt.Errorf("applying static response for pair %d: %w", i, err)
                }
                continue
            }
        }

        // 2. Process the pair
        if err := p.processingStrategy.Process(pair); err != nil {
            return fmt.Errorf("processing pair %d: %w", i, err)
        }
    }

    return nil
}

Let's skip the static endpoint for now and look closer at the processing of each pair.
For prefixed patterns in requests:

func (p *PrefixProcessor) ProcessRequest(value string) (string, bool) {
    if p.replaceWith != "" {
        return p.replaceWith, true
    }

    // Properly escape any regex special characters in the prefix
    escapedPrefix := regexp.QuoteMeta(p.prefix)

    // Check if we detected a resource name embedded in the value
    if resourceName := p.detectResourceName(value); resourceName != "" {
        p.includeResourceName = true
        p.resourceName = resourceName

        // Pattern with resource name included
        return fmt.Sprintf("%s%s-[a-zA-Z0-9]{%d}", escapedPrefix, resourceName, p.randomLength), true
    }

    // Standard pattern without resource name
    return fmt.Sprintf("%s[a-zA-Z0-9]{%d}", escapedPrefix, p.randomLength), true
}

And for responses:

func (p *PrefixProcessor) ProcessResponse(field string, value string, modifiedFields map[string]bool) (string, bool) {
    if !p.Match(value) {
        return value, false
    }

    // If this field was modified in the request and no fixed replacement
    if modifiedFields[field] && p.replaceWith == "" {
        return fmt.Sprintf("{{ Request.Body 'jsonpath' '$.%s' }}", field), true
    }

    if p.replaceWith != "" {
        return p.replaceWith, true
    }

    return value, false
}

This approach:

Changes the matcher for fields to use regex matcher
Uses Hoverfly's templating system to extract values from the incoming request
Uses those same values in the response, maintaining consistency

Following a similar principle, let's look at the UUID example:

func (p *UUIDProcessor) ProcessRequest(_ string) (string, bool) {
    if p.replaceWith != "" {
        return p.replaceWith, true
    }
    // Use proper regex pattern for UUIDs that will work with Hoverfly's RegexMatch
    return `[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}`, true
}

func (p *UUIDProcessor) ProcessResponse(field string, value string, modifiedFields map[string]bool) (string, bool) {
    if !p.Match(value) {
        return value, false
    }

    // If we have a fixed replacement, use it regardless of modified fields
    if p.replaceWith != "" {
        return p.replaceWith, true
    }

    // If this field was modified in the request and modifiedFields is provided
    if modifiedFields != nil && modifiedFields[field] {
        return fmt.Sprintf("{{ Request.Body 'jsonpath' '$.%s' }}", field), true
    }

    return value, false
}

All these processors are orchestrated:

// Process handles both the request and response processing for a pair
func (s *DefaultProcessingStrategy) Process(pair *v2.RequestMatcherResponsePairViewV5) error {
    // First try to process the request
    modifiedFields, err := s.ProcessRequest(pair)
    if err != nil {
        return err
    }

    // Then process the response
    if len(modifiedFields) > 0 {
        // If we modified fields in the request, process response with them
        if err = s.ProcessResponse(pair, modifiedFields); err != nil {
            return err
        }
        return nil
    }

    // If we didn't modify any request fields (e.g., no request body or no matches),
    // try to process the response independently
    if err = s.processResponseForRequestsWithoutBody(pair); err != nil {
        return err
    }

    return nil
}

Here we:

First process the request matchers, identifying dynamic fields
Tracks which fields were modified in the request
Process the response, applying templating where appropriate
Handles special cases like requests without bodies

Note: I understand that the code snippets above can be a bit complex to digest on first reading. Don't worry if you don't grasp all the implementation immediately - they're included primarily for those who want to dive deeper. If you prefer, you can focus on the overall concept rather than the specific code. The full source code is available on GitHub if you want to explore further.

To summarize the main idea: We modify the output JSON files, telling Hoverfly to use the request data in the response. You can, of course, modify this approach to make responses with static data, random data, or whatever else you need.

API Retries

Sometimes things don't work, at least not after the first attempt. Our application might need to handle temporary failures by retrying requests.
Let's look at a test server that simulates an unreliable API:

func (ts *TestServer) handleGetActivity(w http.ResponseWriter, r *http.Request) {
    attempt := ts.attempts.Add(1)

    // Extract the id from the URL path
    pathParts := strings.Split(r.URL.Path, "/")
    if len(pathParts) < 5 {
        http.Error(w, "Invalid URL", http.StatusBadRequest)
        return
    }
    idStr := pathParts[4]
    id, err := strconv.Atoi(idStr)
    if err != nil {
        http.Error(w, "Invalid ID", http.StatusBadRequest)
        return
    }

    // Simulate different responses based on attempt number
    switch attempt {
    case 1:
        // First attempt - Service Unavailable
        w.WriteHeader(http.StatusServiceUnavailable)
        json.NewEncoder(w).Encode(map[string]string{
            "error": "Service Temporarily Unavailable",
            "code":  "SERVER_BUSY",
        })
    case 2:
        // Second attempt - Gateway Timeout
        w.WriteHeader(http.StatusGatewayTimeout)
    case 3:
        // Third attempt - Success
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(http.StatusOK)
        _ = json.NewEncoder(w).Encode(map[string]interface{}{
            "id":        int32(id),
            "title":     "Test Activity",
            "dueDate":   "2025-02-20T09:58:59.009Z",
            "completed": false,
        })
    default:
        // Any subsequent attempts - Bad Request
        w.WriteHeader(http.StatusTooManyRequests)
        _ = json.NewEncoder(w).Encode(map[string]string{
            "error": "Too many attempts",
        })
    }
}

This server:

Returns a 503 (Service Unavailable) on the first attempt
Returns a 504 (Gateway Timeout) on the second attempt
Succeeds with a 200 (OK) on the third attempt

By default, Hoverfly would only capture the final successful response. And of course our test will fail, but like we have a trick for that as well.

Enabling Stateful Capture

To capture this sequence of responses, we need to enable stateful capture mode in Hoverfly:

hoverctl mode capture --stateful

How Stateful Capture Works

When in stateful capture mode, Hoverfly adds metadata to the captured request/response pairs to track their sequence:

requiresState - Added to the request to indicate which state is required to match this entry
transitionsState - Added to the response to indicate how the state should change after this response

Let's look at how this appears in the simulation file:

Here's how the sequence works:

Hoverfly automatically creates a state variable called sequence:1 and initializes it to "1"
The first request matches when the state is "1"
After serving the first response, the state transitions to "2"
The second request matches when the state is "2"
And so on…
After serving the last response, there's no further transition defined

Testing with Stateful Simulations

When we run tests against this simulation, Hoverfly will play back the sequence of responses in order, perfectly simulating our retry scenario:

func TestActivitiesWithRetries(t *testing.T) {
    serverUrl, ok := os.LookupEnv("API_SERVER_URL")
    if !ok || serverUrl == "" {
        ts := tests.NewTestServer() // Start our test server
        defer ts.Close()
        serverUrl = ts.URL()
    }

    httpClient := tests.NewHttpClient(t)
    client, err := NewClientWithResponses(serverUrl, WithHTTPClient(httpClient))
    if err != nil {
        t.Fatalf("failed to inistialize client: %v", err)
    }
    defer func() {
        // Reset the server state
        _, _ = httpClient.Post(serverUrl+"/reset", "application/json", nil)
    }()

    scenarios := []testCase{
        {
            name: "create activity with retries",
            testFunc: func(t *testing.T, client *ClientWithResponses) {
                var (
                    ctx   = context.Background()
                    id    = int32(7474)
                    title = "Test Activity"
                )

                var lastErr error
                for attempt := 1; attempt <= 4; attempt++ {
                    resp, err := client.GetApiV1ActivitiesIdWithResponse(
                        ctx,
                        id,
                    )
                    if err != nil {
                        lastErr = err
                        continue
                    }

                    // Success!
                    if resp.StatusCode() == http.StatusOK {
                        require.NotNil(t, resp.ApplicationjsonV10200)
                        assert.Equal(t, id, *resp.ApplicationjsonV10200.Id)
                        assert.Equal(t, title, *resp.ApplicationjsonV10200.Title)
                        return
                    }

                    // Continue if it's a retryable status
                    if resp.StatusCode() == http.StatusServiceUnavailable ||
                        resp.StatusCode() == http.StatusGatewayTimeout {
                        continue
                    }

                    // Non-retryable error
                    t.Fatalf("Got non-retryable status: %d", resp.StatusCode())
                }

                // If we got here, all retries failed
                t.Fatalf("All retries failed. Last error: %v", lastErr)
            },
        },
    }

    for _, tt := range scenarios {
        t.Run(tt.name, func(t *testing.T) {
            tt.testFunc(t, client)
        })
    }
}

Benefits of Stateful Simulation

Using stateful capture may provide several benefits:

Test retry mechanisms properly - Verify that your client correctly handles temporary errors
Simulate workflows - Test multi-step processes that change state with each request
Reproduce edge cases - Capture and replay rare error conditions

By combining our pattern processors with stateful capture, we gain the ability to test both dynamic data and complex stateful interactions without changing our test code.

Static Response Replacement

Sometimes you want to provide a completely different response for specific endpoints. This is useful for:

Creating test-specific behavior - Returning predefined responses for certain test cases
Handling edge cases - Simulating error conditions or special scenarios
Maintaining consistent test data - Ensuring tests run with known values

Our processor supports this through the endpoints section in the configuration:

endpoints:
  - method: "GET"
    path: "/api/v1/Activities/30"
    status: 200
    static_response: |
      {
        "id": 77,
        "title": "John Doe",
        "dueDate": "2025-02-19T17:12:52.127Z",
        "completed": false
      }

This completely overrides any captured response for the specified endpoint, giving you precise control over the behavior in your tests.

Conclusion

We've explored several powerful techniques for API testing with simulations:

Understanding Hoverfly's matching system - Learning how request matchers work
Handling dynamic data - Making simulations work with UUIDs, random strings, and timestamps
Testing retry logic - Using stateful capture to simulate unreliable APIs
Static response replacement - Overriding specific endpoints for test scenarios

Of course, this approach may not be for everyone. If you require much more flexibility or you're starting a new project, then tools like WireMock might be a better choice. But there are specific use cases where this approach is particularly helpful and more effective than alternatives, making Hoverfly a useful tool to have in your testing arsenal.

In the final part of this series, we'll explore integrating these tools into CI pipelines and development workflows.

Source Code

The complete source code for this solution is available on GitHub.

API testing through simulations

Myroslav Vivcharyk — Sun, 02 Mar 2025 14:36:41 +0000

The Challenge of API Testing

API testing is a challenging aspect of software development. While there are many approaches available — from mocking to contract testing — each comes with its own set of trade-offs.
Rather than debating the pros and cons of different testing approaches (feel free to draw your testing pyramids in the comments), this series of articles focuses on a specific scenario: testing against real APIs. Whether it’s production, staging, or a sandbox environment, we’ll explore how to improve tests that interact with actual API endpoints.

The Real-World Context

Let’s face it: many development teams find themselves deeply integrated with real APIs — and that’s normal. Instead of suggesting a complete architectural overhaul or debating how we arrived at this point, let’s focus on practical improvements to your existing setup.
When interacting with real APIs, developers often face several critical challenges:

Network Reliability: Random connection errors and timeouts leading to flaky tests
Latency: API calls taking seconds (or even minutes!) to complete, significantly slowing down test execution
Resource Consumption: Every test run potentially consuming API quotas or incurring actual costs
Environment Stability: External API availability and rate limits affecting test reliability

A Practical Solution

We’ll explore Hoverfly — an open-source API simulation tool that deserves more attention. Despite its capabilities, comprehensive examples and real-world use cases can be hard to find — which is precisely why we’re writing this series.

“Hoverfly is a lightweight, open source API simulation tool. Using Hoverfly, you can create realistic simulations of the APIs your application depends on.”

What does this mean in practice? We can run our tests against a real API once (or periodically), record all interactions, and use these recordings for future test runs. This gives us the benefits of both worlds: real API behavior with predictable test data.

💡 Want to dive straight into the code? Check out the complete source code here.

Prerequisites

Docker and Docker Compose
Go 1.23 or later
Task (taskfile) for running our commands

Setting Up Our Test Environment

First, let’s set up our development environment. We’ll use OpenAPI Generator within Docker to generate our HTTP client, keeping our local environment clean. For this demonstration, we’ll work with fakerestapi.azurewebsites.net (kudos to the maintainers of public APIs).
Here’s our Docker configuration for the client generator:

FROM golang:1.23-alpine

RUN apk add --no-cache git

RUN go install github.com/oapi-codegen/oapi-codegen/v2/cmd/oapi-codegen@v2.4.1

WORKDIR /app
ENTRYPOINT ["oapi-codegen"]

And the corresponding docker-compose service:

services:
  oapi-generator:
    container_name: oapi-generator
    build:
      context: .
      dockerfile: ./docker/oapi.Dockerfile
    volumes:
      - .:/app
    working_dir: /app
    command:
      - "-package"
      - "oapi"
      - "-generate"
      - "types,client"
      - "-o"
      - "./client/oapi/client.go"
      - "./api/test_api.json"

Writing First Tests

With our client generated, let’s write some basic tests to demonstrate API interactions. We’ll focus on the Authors endpoint for this example (but you can more in the source):

func TestAuthors(t *testing.T) {
 scenarios := []testCase{
  {
   name: "get author",
   testFunc: func(t *testing.T, client *ClientWithResponses) {
    var (
     ctx = context.Background()
     id  = int32(1)
    )

    resp, err := client.GetApiV1AuthorsIdWithResponse(ctx, id)
    require.NoError(t, err)
    require.NotNil(t, resp, "expected resp, got nil")
    require.NotNil(t, resp.ApplicationjsonV10200, "expected resp body, got nil")

    assert.Equal(t, http.StatusOK, resp.StatusCode())
    assert.Equal(t, id, *resp.ApplicationjsonV10200.Id)
    assert.NotEmpty(t, *resp.ApplicationjsonV10200.FirstName, "expected non-empty first name")
    assert.NotEmpty(t, *resp.ApplicationjsonV10200.LastName, "expected non-empty last name")
   },
  },
  {
   name: "create new author",
   testFunc: func(t *testing.T, client *ClientWithResponses) {
    var (
     ctx       = context.Background()
     id        = int32(1)
     firstName = randomTitleForResource("author")
     lastName  = randomTitleForResource("author")
    )

    resp, err := client.PostApiV1AuthorsWithApplicationJSONV10BodyWithResponse(
     ctx,
     PostApiV1AuthorsApplicationJSONV10RequestBody{
      Id:        tests.ToPtr(id),
      FirstName: tests.ToPtr(firstName),
      LastName:  tests.ToPtr(lastName),
     },
    )
    require.NoError(t, err)
    require.NotNil(t, resp, "expected resp, got nil")
    require.NotNil(t, resp.ApplicationjsonV10200, "expected resp body, got nil")

    assert.Equal(t, http.StatusOK, resp.StatusCode())
    assert.Equal(t, id, *resp.ApplicationjsonV10200.Id)
    assert.Equal(t, firstName, *resp.ApplicationjsonV10200.FirstName)
    assert.Equal(t, lastName, *resp.ApplicationjsonV10200.LastName)
   },
  },
 }

 client := setupClient(t)

 for _, tt := range scenarios {
  t.Run(tt.name, func(t *testing.T) {
   tt.testFunc(t, client)
  })
 }
}

func setupClient(t *testing.T) *ClientWithResponses {
 httpClient := http.DefaultClient
 client, err := NewClientWithResponses(baseURL, WithHTTPClient(httpClient))
 if err != nil {
  t.Fatalf("failed to inistialize client: %v", err)
 }

 return client
}

Enter Hoverfly: Setting Up API Simulation

The core concept in Hoverfly is “simulations” — JSON files that store request-response pairs from your API interactions. Let’s set up Hoverfly in a Docker container:

FROM alpine:3.21.0

# Packages
RUN apk add --no-cache wget unzip curl

# Set default arguments
ARG HOVERFLY_VERSION="v1.10.6"
ARG HOVERFLY_ADMIN_PORT=8888
ARG HOVERFLY_PROXY_PORT=8500

ENV HOVERFLY_ADMIN_PORT=${HOVERFLY_ADMIN_PORT}
ENV HOVERFLY_PROXY_PORT=${HOVERFLY_PROXY_PORT}

# Download and install both hoverfly and hoverctl
RUN wget -q "https://github.com/SpectoLabs/hoverfly/releases/download/v${HOVERFLY_VERSION#v}/hoverfly_bundle_linux_amd64.zip" && \
    unzip hoverfly_bundle_linux_amd64.zip -d /tmp && \
    mv /tmp/hoverfly /usr/local/bin/ && \
    mv /tmp/hoverctl /usr/local/bin/ && \
    chmod +x /usr/local/bin/hoverfly && \
    chmod +x /usr/local/bin/hoverctl && \
    rm -rf hoverfly_bundle_linux_amd64.zip /tmp/*

# Create default hoverctl config with environment variables
RUN mkdir -p /root/.hoverfly && \
    echo "hoverfly.host: localhost" > /root/.hoverfly/config.yaml && \
    echo "hoverfly.admin.port: \"${HOVERFLY_ADMIN_PORT}\"" >> /root/.hoverfly/config.yaml && \
    echo "hoverfly.proxy.port: \"${HOVERFLY_PROXY_PORT}\"" >> /root/.hoverfly/config.yaml

EXPOSE ${HOVERFLY_PROXY_PORT} ${HOVERFLY_ADMIN_PORT}

ENTRYPOINT ["hoverfly", "-listen-on-host=0.0.0.0"]

Add new docker-compose service to our configuration:

  hoverfly:
    container_name: api-test-demo-hoverfly
    build:
      context: ./docker
      dockerfile: hoverfly.Dockerfile
      args:
        - HOVERFLY_VERSION=1.10.6
        - HOVERFLY_ADMIN_PORT=${HOVERFLY_ADMIN_PORT:-8888}
        - HOVERFLY_PROXY_PORT=${HOVERFLY_PROXY_PORT:-8500}
    ports:
      - "${HOVERFLY_HOST_ADMIN_PORT:-8888}:${HOVERFLY_ADMIN_PORT:-8888}"
      - "${HOVERFLY_HOST_PROXY_PORT:-8500}:${HOVERFLY_PROXY_PORT:-8500}"
    volumes:
      - ./testdata/hoverfly:/testdata/hoverfly
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:${HOVERFLY_ADMIN_PORT:-8888}/api/health"]
      interval: 1s
      timeout: 3s
      retries: 5

This setup provides:

A Hoverfly instance
Web UI for monitoring (port 8888)
Proxy server for API interception (port 8500)
Persistent storage for our simulations The last step would be adding proxy for our HTTP client configuration. For this we need to add a tiny change:

func NewHttpClient(t *testing.T) *http.Client {
    t.Helper()

    hoverflyAddr := os.Getenv("HOVERFLY_PROXY")
    if hoverflyAddr == "" {
        return http.DefaultClient
    }

    // We will run tests in parallel only if we are using Hoverfly.
    // Just to avoid redundant load for public API.
    t.Parallel()

    client := &http.Client{
        Timeout: 5 * time.Second,
        Transport: &http.Transport{
            Proxy: http.ProxyURL(&url.URL{
                Scheme: "http",
                Host:   hoverflyAddr,
            }),
            TLSClientConfig: &tls.Config{
                InsecureSkipVerify: true, // Skip certificate verification when using Hoverfly
            },
        },
    }

    return client
}

Recording and Replaying API Interactions

Using task definitions, we can capture API interactions:

test:local:capture:
    desc: Run tests on local machine and capture traffic with Hoverfly. Export simulation to file.
    env:
      HOVERFLY_PROXY: '{{.HOVERFLY_PROXY}}'
    vars:
      TIMESTAMP:
        sh: date +%Y%m%d_%H%M%S
    cmds:
      - defer: {task: hoverfly:stop}
      - task: internal:bootstrap
        vars:
          CLI_ARGS: hoverfly
      - task: hoverfly:mode
        vars:
          MODE: capture
      - task: test:local
        vars:
          HOVERFLY_PROXY: "localhost:{{.HOVERFLY_PORT | default 8500}}"
      - task: hoverfly:export
        vars:
          SIMULATIONS_DIR: '{{.SIMULATIONS_DIR}}'
          SIMULATION_FILE: '{{.TIMESTAMP}}'

   test:local:
    desc: |
      Run all tests on local machine. By default, tests are run against real API.
      Optionally use Hoverfly for simulation or capturing by setting HOVERFLY_PROXY environment variable and running hoverfly service.
    deps:
      - internal:check-gotestsum
    cmds:
      - PATH="$(go env GOPATH)/bin:$PATH" gotestsum
        --format pkgname
        --hide-summary=skipped
        --
        -json ./...
        -race
        -timeout=30s
        -count=1
    env:
      HOVERFLY_PROXY: '{{.HOVERFLY_PROXY | default ""}}'

When we run this task, it:

Starts Hoverfly container
Enables capture mode
Runs tests through the proxy
Saves interactions to a JSON file

Running Simulations

Once we’ve captured our API interactions, we can run tests using the simulated responses. Here’s our task definition:

test:local:simulate:
    desc: Run tests locally simulating the API with Hoverfly
    deps:
      - task: internal:bootstrap
        vars:
          CLI_ARGS: hoverfly
    vars:
      LATEST_SIM:
        sh: |
          if ! SIM=$(task internal:find:latest:simulation SIMULATIONS_DIR={{.SIMULATIONS_DIR}} EXPIRED_AFTER_DAYS={{.EXPIRED_AFTER_DAYS}}); then
            echo ""
          else
            echo "$SIM"
          fi
    preconditions:
      - sh: '[ -f "{{.LATEST_SIM}}" ]'
        msg: "Simulation file does not exist. Please run 'task test:local:capture' first"
    cmds:
      - task: hoverfly:mode
        vars:
          MODE: simulate
      - task: hoverfly:import
        vars:
          SIMULATION_FILE: '{{.LATEST_SIM}}'
      - task: test:local
        vars:
          HOVERFLY_PROXY: "localhost:{{.HOVERFLY_PORT | default 8500}}"

The workflow is straightforward:

Hoverfly starts in simulate mode
We load our previously captured simulation file (the task automatically finds the most recent one)
Tests run against Hoverfly instead of the real API

Validating Our Setup

To verify that our simulation is actually working (and not secretly hitting the real API), let’s intentionally break one of our tests:

func TestAuthors(t *testing.T) {
    scenarios := []testCase{
        {
            name: "get author",
            testFunc: func(t *testing.T, client *ClientWithResponses) {
                var (
                    ctx = context.Background()
                    id  = int32(1)
                )

                resp, err := client.GetApiV1AuthorsIdWithResponse(ctx, id)
                require.NoError(t, err)
                require.NotNil(t, resp, "expected resp, got nil")
                require.NotNil(t, resp.ApplicationjsonV10200, "expected resp body, got nil")

                assert.Equal(t, http.StatusOK, resp.StatusCode())

                // Intentionally incorrect assertion - the real API returns id=1
                assert.Equal(t, 10, *resp.ApplicationjsonV10200.Id)
            },
        },
    }
}

This failure confirms three points:

Isolation: Tests are running against simulated responses
Accuracy: Simulations faithfully reflect real API behavior
Validation: Test assertions work correctly against simulated data

Summary

Let’s recap what we’ve built: a testing infrastructure that combines real API testing with simulation capabilities. The best part? We achieved this with minimal changes to our existing codebase. No need for extensive refactoring or rewriting tests — we simply added a proxy configuration to our HTTP client and let Hoverfly handle the rest.
The workflow is straightforward — we write our tests as usual, but run them through a Hoverfly proxy server. Hoverfly captures API interactions and stores them as simulation files. For subsequent test runs, Hoverfly can replay these stored responses instead of hitting the real API.
This setup gives us the benefits of both worlds — we’re testing against real API behavior, but with the reliability of local tests. No more flaky tests due to network issues, no more burning through API quotas during development, and most importantly, consistent test execution across all environments.

💡 A Note on Performance Expectations

While Hoverfly is excellent for reliability and simulation, it’s important to manage your expectations. If your API typically responds within 150–200ms, you might not see significant speed improvements in your tests.
Choose Hoverfly for reliability and consistent behavior first, speed improvements second.

Looking Ahead

In the next part of this series, we’ll explore:

Dynamic Data: Handling variable responses
CI/CD Integration: Seamless pipeline implementation