Drain Pattern — Professional Level¶

Table of Contents¶

1. Introduction
2. Kafka Consumer Architecture Recap
3. The Rebalance Protocol
4. Drain Hooks: onPartitionsRevoked
5. Exactly-Once Drain
6. Cooperative Rebalance and Drain
7. Drain Across Worker Pool Backed By Kafka
8. Idempotent Producer Drain
9. Transactional Producer Drain
10. Drain Across Process Boundaries
11. War Stories — Drain Incidents In Production
12. Drain In Distributed Pipelines
13. Drain And Exactly-Once Theory
14. Drain Performance Engineering
15. Drain Observability At Scale
16. Drain Frameworks Comparison
17. Drain Across Cloud Vendors
18. Drain Anti-Patterns At Professional Level
19. Drain Lessons That Took Years To Learn
20. Summary

Introduction¶

Focus: "Drain Kafka consumers without dropping or duplicating a single message. Drain transactional producers without breaking exactly-once. Drain across an entire stream-processing pipeline so a rolling deploy is invisible to downstream."

This is the deep end. Junior, middle, and senior covered the patterns in increasing breadth. Professional covers the specific, demanding scenario where every bit of drain matters: Kafka consumers in exactly-once pipelines, where a single dropped offset is a duplicate financial event in production. The patterns from earlier pages are the foundation; this page is the application.

After reading this file you will:

• Understand Kafka's rebalance protocol and how drain integrates with it.
• Implement onPartitionsRevoked hooks that commit and clean up cleanly.
• Drain an exactly-once consumer with transactional commits.
• Coordinate drain across producers and consumers in the same pipeline.
• Recognise the failure modes specific to drain in messaging systems.
• Apply the lessons across non-Kafka systems with similar semantics (Pulsar, Kinesis, NATS JetStream).

Kafka Consumer Architecture Recap¶

A Kafka consumer is a goroutine that reads messages from one or more partitions. Partitions are units of parallelism. A consumer group has many consumers; the group coordinator (a broker) assigns partitions to consumers. The assignment can change at any time — when consumers join, leave, or are deemed unhealthy. This is a rebalance.

A consumer's lifecycle within a group:

1. Joins the group.
2. Receives partition assignments.
3. Fetches messages from each assigned partition.
4. Processes each message.
5. Periodically commits offsets (__consumer_offsets topic).
6. Possibly loses partitions during rebalance.
7. Eventually leaves the group.

Drain interacts with steps 4, 5, and 6 in particular. A drain that does not commit offsets cleanly causes duplicates on the next consumer.

Drain modes¶

• Static drain. Process gracefully exits. Group coordinator notices, triggers rebalance.
• Live rebalance. Other consumer joins or leaves. Coordinator triggers rebalance; this consumer loses some partitions.

Both must be handled. Static drain is what SIGTERM triggers; live rebalance is what scale events trigger.

The Rebalance Protocol¶

Kafka's rebalance has historically been "stop the world": when partitions are reassigned, all consumers stop processing, give up all partitions, and receive new assignments. The pause can be seconds.

Modern Kafka (since 2.4) supports cooperative incremental rebalance: only the partitions that are moving stop processing; others continue. The pause is per-partition, not global.

Either way, the consumer has hooks for the transition:

• onPartitionsRevoked(partitions) — before partitions are taken away.
• onPartitionsAssigned(partitions) — after new partitions arrive.
• onPartitionsLost(partitions) — when partitions were taken away abruptly (rare).

Drain happens inside onPartitionsRevoked. This is the one chance to commit offsets, flush in-flight processing, and release resources for the partitions being lost.

Drain Hooks: onPartitionsRevoked¶

Pseudocode for a drain-aware consumer:

type ConsumerGroupHandler struct {
    inFlight map[int32]*sync.WaitGroup // partition -> wg
    commits  map[int32]int64           // partition -> offset
    mu       sync.Mutex
}

func (h *ConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim) error {
    p := claim.Partition()
    wg := h.wgForPartition(p)
    for msg := range claim.Messages() {
        wg.Add(1)
        go func(m *sarama.ConsumerMessage) {
            defer wg.Done()
            if err := process(m); err != nil {
                // handle
                return
            }
            h.mu.Lock()
            h.commits[m.Partition] = m.Offset + 1
            h.mu.Unlock()
        }(msg)
    }
    return nil
}

func (h *ConsumerGroupHandler) Cleanup(sess sarama.ConsumerGroupSession) error {
    // Called when partitions are about to be revoked.
    for p, wg := range h.inFlight {
        wg.Wait()
        h.mu.Lock()
        offset := h.commits[p]
        h.mu.Unlock()
        sess.MarkOffset(claim.Topic(), p, offset, "")
    }
    sess.Commit()
    return nil
}

The pattern:

• Track in-flight per partition.
• On revoke, wait for that partition's in-flight to finish.
• Commit the highest committed offset.
• Then let the rebalance proceed.

A naive consumer that does not wait in Cleanup causes duplicates: messages still being processed when the partition is reassigned will be re-delivered to the new owner.

Exactly-Once Drain¶

Exactly-once delivery in Kafka requires:

• Idempotent producer.
• Transactional commit of (output messages + input offsets).
• Read-committed isolation on downstream consumers.

Drain interacts with the transaction:

• A transaction in-flight at drain start must either commit or abort, atomically.
• Aborting loses the work; committing requires the output messages to be safely produced.

The pattern:

producer.BeginTransaction()
process(input)
producer.Produce(output)
producer.SendOffsetsToTransaction(inputOffsets, group)
producer.CommitTransaction()

On drain mid-transaction:

if producer.InTransaction() {
    producer.AbortTransaction()
}

Abort discards both the input offset and the output messages. Next consumer re-reads input and produces output anew.

For long transactions, the drain budget must accommodate the transaction's longest reasonable duration. If a transaction takes 30s and drain budget is 25s, you have a structural mismatch.

Cooperative Rebalance and Drain¶

Cooperative rebalance changed the drain story significantly. With eager rebalance, drain meant "stop everything, commit everything, restart everything." With cooperative rebalance, drain can be partial: only the partitions being revoked stop; others keep running.

Implications:

• The drain pattern moves from "global pause" to "per-partition drain."
• Each partition has its own in-flight count and its own commit point.
• The supervisor must track per-partition state.

type PartitionState struct {
    wg          sync.WaitGroup
    committed   atomic.Int64
    processed   atomic.Int64
    revoking    atomic.Bool
}

type Consumer struct {
    partitions sync.Map // map[int32]*PartitionState
}

func (c *Consumer) OnPartitionsRevoked(ps []int32) {
    for _, p := range ps {
        state := c.getOrCreate(p)
        state.revoking.Store(true)
        state.wg.Wait() // bounded by Kafka client's rebalance timeout
        c.commit(p, state.committed.Load())
    }
}

Cooperative rebalance still has a deadline — the broker side session.timeout.ms. Drain must complete inside it.

Drain Across Worker Pool Backed By Kafka¶

A common architecture: a consumer feeds messages into a worker pool; workers process and commit. Drain order:

1. Stop the consumer's fetch.
2. Drain the worker pool (wait for in-flight workers).
3. Commit offsets for all processed messages.
4. Disconnect from Kafka.

type Pipeline struct {
    consumer *Consumer
    pool     *WorkerPool
    tracker  *OffsetTracker
}

func (p *Pipeline) Drain(ctx context.Context) error {
    // Step 1: stop consumer fetch.
    if err := p.consumer.StopFetch(ctx); err != nil {
        return err
    }
    // Step 2: drain workers.
    if err := p.pool.Drain(ctx); err != nil {
        return err
    }
    // Step 3: commit highest processed offset per partition.
    if err := p.tracker.CommitAll(ctx); err != nil {
        return err
    }
    // Step 4: disconnect.
    return p.consumer.Close()
}

The order is critical. Disconnect before commit and the commit fails. Commit before drain and you commit offsets the worker has not actually processed.

Idempotent Producer Drain¶

An idempotent producer attaches a producer ID + sequence number to each message. Duplicates are deduped server-side.

Drain semantics for idempotent producer:

• Pending messages must be flushed before close.
• Failed sends should be retried (within bounds).
• The producer's internal queue must drain.

func (p *Producer) Drain(ctx context.Context) error {
    if err := p.client.Flush(ctx); err != nil {
        return err
    }
    return p.client.Close()
}

Flush blocks until all pending messages are acknowledged. With a context, the wait is bounded.

If Flush times out, the unflushed messages are lost. The application should either:

• Increase the deadline (if reasonable).
• Persist the messages locally for retry on restart.
• Accept the loss (for non-critical telemetry).

Transactional Producer Drain¶

A transactional producer wraps idempotent semantics with multi-message atomicity. Drain considerations:

• If a transaction is in flight, decide commit or abort.
• If committing, ensure all messages and offsets are sent.
• If aborting, the work is discarded.

func (p *TxProducer) Drain(ctx context.Context) error {
    if p.InTransaction() {
        // Choice: commit or abort?
        if err := p.CommitTransaction(ctx); err != nil {
            _ = p.AbortTransaction(ctx)
            return err
        }
    }
    return p.client.Close()
}

Choosing commit-over-abort during drain:

• Commit if the transaction is short and likely to succeed.
• Abort if the transaction is large or partially failed.

The pattern most often chosen: commit if BeginTransaction was called and no errors have occurred; abort otherwise.

Drain Across Process Boundaries¶

A stream processing pipeline often spans many processes:

• Producer service (Go).
• Kafka cluster (Java).
• Consumer/transformer (Go).
• Downstream consumer (Python, Go, or other).

Drain semantics across the pipeline:

• Each Go service drains via signal.
• Kafka brokers handle rebalance.
• Downstream services may not even know an upstream is draining.

The pipeline's overall consistency depends on each component's correctness. A drained upstream leaves no in-flight messages; a stopping-without-drain upstream may leave the pipeline in a transient inconsistent state.

For exactly-once pipelines, every component must support drain. A single non-drain-aware component is a single point of failure.

War Stories — Drain Incidents In Production¶

Story 1: The Tuesday Outage¶

A team rolled out a new version of their order-processing service. The drain code was untested. On SIGTERM, the consumer abruptly closed Kafka connections without committing offsets. 47 minutes of duplicate orders before the issue was noticed. Customer support spent the week refunding double charges.

Root cause: missing Cleanup implementation on the ConsumerGroupHandler. The default was a no-op. The team assumed the Kafka client handled drain. It did not.

Fix: implement Cleanup with explicit commit-and-wait. Add a drain test in CI.

Story 2: The Slow Drain Mystery¶

A service was reliable for a year. Then drain times started creeping up: 5s, 8s, 12s. Eventually drain hit the 30s grace period and pods started SIGKILL-ing.

Root cause: a goroutine leak in a third-party tracing library. Every request leaked a goroutine that held a reference to the request body. After a week of uptime, drain wait time was dominated by these leaked goroutines.

Fix: upgrade the tracing library. Add a goroutine count metric. Add an alert.

Story 3: The Partition Skew¶

A consumer service had 12 partitions across 6 consumers, balanced 2-2. A rolling deploy started: consumer 1 drained, consumer 2 took over its partitions. But consumer 2 still had its original 2 partitions in flight when the assignment changed. It briefly had 4 partitions to drain — 2x its capacity.

The drain took 60 seconds. Pods SIGKILL-ed. 30 minutes of message backlog.

Root cause: drain deadline assumed steady-state in-flight count, not the spike during rebalance.

Fix: budget drain time for rebalance-induced spikes. Reduce per-partition work. Use cooperative rebalance.

Story 4: The Transaction Trap¶

A team enabled exactly-once on a service. Drain logic was unchanged. On the first deploy after enabling EOS, drain caused transaction aborts. 10% of transactions discarded their work.

Root cause: drain code aborted in-flight transactions. The team had not changed this when enabling EOS.

Fix: check transaction state in drain; commit if possible, abort if not.

Story 5: The Cascading Cancel¶

A service's drain context was passed to all downstream calls. A downstream call timed out at 30s; the drain deadline was 25s. When drain started, the call's context was already cancelled. All in-flight calls failed simultaneously.

Root cause: drain context too aggressive. The downstream had not been told about drain.

Fix: separate drain context from request context. Use shorter sub-deadlines per call.

Drain In Distributed Pipelines¶

A distributed pipeline (Apache Flink-like, Kafka Streams-like) drains stage by stage:

1. Source connector stops fetching.
2. First stage drains.
3. Second stage drains.
4. Sink connector commits final state.

Each stage's drain depends on the previous. The pipeline's drain time is the sum of stage drain times.

For Go pipelines built on Kafka:

type Stage interface {
    Process(ctx context.Context, in <-chan Message, out chan<- Message) error
    Drain(ctx context.Context) error
}

type Pipeline struct {
    stages []Stage
}

func (p *Pipeline) Drain(ctx context.Context) error {
    for _, s := range p.stages {
        if err := s.Drain(ctx); err != nil {
            return err
        }
    }
    return nil
}

Sequential drain. Total time is the sum. For long pipelines, this can exceed grace period; mitigations:

• Drain stages in parallel where state is independent.
• Reduce per-stage work.
• Increase grace period.

Drain And Exactly-Once Theory¶

Exactly-once semantics (EOS) in distributed systems requires:

• Idempotent operations.
• Atomic commit of effects.
• Recovery from failure that respects atomicity.

Drain is one source of "failure": a process exiting voluntarily. If drain mid-transaction is treated as a crash, the system must recover correctly.

Properly designed:

• Transactions are bounded (sub-second to a few seconds).
• Each transaction is independent.
• Recovery replays from the last committed transaction.
• Drain attempts to complete in-flight transactions; aborts otherwise.

EOS is hard. Drain in EOS is harder. The senior + professional levels of drain are mostly about not breaking EOS.

Drain Performance Engineering¶

For high-throughput services, drain becomes a performance problem:

• 100k messages/sec consumer. 10s drain. 1M messages in-flight.
• Each message has its own goroutine. 1M goroutines.
• Drain wait group has 1M entries.

Performance considerations:

• WaitGroup operations are atomic; at 1M concurrent, the cache line bounces.
• Goroutine stacks consume memory (2KB each = 2GB for 1M).
• GC pressure during drain spikes.

Mitigations:

• Use a worker pool with a bounded queue, not goroutine-per-message.
• Track in-flight via a sharded counter.
• Pre-allocate buffers; reuse via sync.Pool.
• Test drain at peak throughput, not at idle.

A drain that is fast at low load and slow at peak load is a drain bug waiting to be a production incident.

Drain Observability At Scale¶

For a fleet of 1000 pods, drain observability requires:

• Per-pod drain duration metric.
• Aggregate drain duration histogram (across fleet).
• Drain failures counter.
• Drain force-cancellation counter.
• Goroutines-at-drain-start gauge.
• Memory-at-drain-start gauge.

Alert on:

• P99 drain duration > 80% of grace period for 5 min.
• Force-cancellation rate > 0.1% of drains.
• Drain failures > 0.

Dashboards:

• Drain duration over time, by service.
• Drain duration vs version (catch regressions).
• Drain duration vs load.
• Goroutine count delta during drain.

At fleet scale, the dashboards become essential. Individual pod inspection is impossible.

Drain Frameworks Comparison¶

For new projects, franz-go or confluent-kafka-go are the strongest. Both support drain via standard Go patterns.

Drain Across Cloud Vendors¶

• AWS MSK / Kinesis. Drain is similar to Kafka. Kinesis has different semantics; the shard model affects drain order.
• GCP Pub/Sub. Drain is subscription.Receive ending. Acks must complete.
• Azure Service Bus. Drain via Close on the client; pending receives finish.
• NATS JetStream. Subscription.Drain() is the canonical primitive.
• Apache Pulsar. Consumer.Close() includes drain semantics.

Each cloud has its own naming and quirks. The principle (stop intake, wait, commit, close) is universal.

Drain Anti-Patterns At Professional Level¶

Anti-pattern: Disabling drain to "fix" a deploy¶

A deploy fails. The team disables the drain test "temporarily" to ship. The disable lasts months. Drain rots.

Mitigation: never disable drain tests. If failing, fix the drain.

Anti-pattern: Drain that depends on a remote service being up¶

Drain code that calls a downstream during drain may fail if the downstream is also draining or unreachable.

Mitigation: drain should be local. Communicate drain via in-process state.

Anti-pattern: Skipping commit on drain timeout¶

If drain times out, the team's code may skip the commit "to avoid blocking longer." Result: duplicate processing on next consumer.

Mitigation: always commit what you can, even partial. Better to commit some than none.

Anti-pattern: Drain order coupling¶

Component A's drain reaches into component B. Component B's drain reaches into A. Cycle.

Mitigation: each component drains itself only. Supervisor coordinates.

Anti-pattern: Drain via "kill the goroutine"¶

There is no "kill goroutine" in Go. Code that pretends to kill goroutines (via panics or other tricks) leaks state.

Mitigation: use cooperative cancellation via context.

Anti-pattern: Drain budget without margin¶

Drain budget exactly equals grace period. A slight slowdown causes SIGKILL.

Mitigation: always leave 5-10s margin.

Anti-pattern: Untested drain¶

The most common anti-pattern. Drain code that has never been triggered.

Mitigation: drain test in CI as a deploy gate.

Drain Lessons That Took Years To Learn¶

A list of hard-won lessons:

1. Drain budget is a function of P99 latency under load. Not P50.
2. Readiness propagation matters more than people think. A 2s sleep saves 10s of drain time.
3. Goroutine leaks make drain slower over time. Track and alert.
4. Drain order depends on dependency direction, not on construction order. Reverse-construction is a heuristic; verify against dependencies.
5. Drain context must be fresh, not derived from cancelled parent. This is the #1 drain bug.
6. Drain tests catch more bugs than any other test. Invest in them.
7. Drain is the cheapest reliability investment. ROI is enormous.
8. Drain quality predicts overall engineering quality. Use it as a barometer.
9. Drain failures are interesting incidents. They reveal architectural assumptions.
10. Drain teaching scales. One senior who teaches drain can lift a team.

Summary¶

Drain at the professional level is about specific, high-stakes scenarios: Kafka exactly-once, transactional producers, distributed pipelines. The patterns from earlier pages are the foundation; this page is the demanding application.

If you have absorbed this, you can:

• Design drain into a new Kafka consumer service.
• Audit an existing service for drain bugs.
• Lead an incident review on a drain failure.
• Mentor a team into drain-quality discipline.
• Evaluate frameworks and libraries for drain support.
• Advocate at the org level for drain quality.

The remaining files (specification, interview, tasks, find-bug, optimize) provide reference, practice, and assessment. Use them to consolidate.

Drain is now part of your toolkit. Go build systems that drain well.

Extended Section: A Real Kafka Drain Implementation¶

Below is a more complete, working Kafka consumer drain implementation using segmentio/kafka-go. It demonstrates the patterns from this page in code.

package main

import (
    "context"
    "encoding/json"
    "errors"
    "log"
    "os"
    "os/signal"
    "sync"
    "sync/atomic"
    "syscall"
    "time"

    "github.com/segmentio/kafka-go"
)

type Event struct {
    ID   string `json:"id"`
    Body string `json:"body"`
}

type Consumer struct {
    reader      *kafka.Reader
    processed   sync.Map // map[string]struct{} for idempotency
    inFlight    sync.WaitGroup
    draining    atomic.Bool
    fetchCancel context.CancelFunc
    fetchCtx    context.Context
    parentCtx   context.Context
}

func NewConsumer(brokers []string, topic, group string) *Consumer {
    return &Consumer{
        reader: kafka.NewReader(kafka.ReaderConfig{
            Brokers: brokers,
            GroupID: group,
            Topic:   topic,
            MinBytes: 10e3,
            MaxBytes: 10e6,
        }),
    }
}

func (c *Consumer) Start(ctx context.Context, workers int) {
    c.parentCtx = ctx
    c.fetchCtx, c.fetchCancel = context.WithCancel(ctx)

    msgs := make(chan kafka.Message, 64)

    // fetcher
    go func() {
        defer close(msgs)
        for {
            m, err := c.reader.FetchMessage(c.fetchCtx)
            if err != nil {
                if errors.Is(err, context.Canceled) {
                    return
                }
                log.Printf("fetch: %v", err)
                continue
            }
            select {
            case <-c.fetchCtx.Done():
                return
            case msgs <- m:
            }
        }
    }()

    // workers
    for i := 0; i < workers; i++ {
        c.inFlight.Add(1)
        go c.worker(i, msgs)
    }
}

func (c *Consumer) worker(id int, msgs <-chan kafka.Message) {
    defer c.inFlight.Done()
    for m := range msgs {
        if err := c.process(c.parentCtx, m); err != nil {
            log.Printf("worker %d: process: %v", id, err)
            continue
        }
        if err := c.reader.CommitMessages(c.parentCtx, m); err != nil {
            log.Printf("worker %d: commit: %v", id, err)
        }
    }
}

func (c *Consumer) process(ctx context.Context, m kafka.Message) error {
    var e Event
    if err := json.Unmarshal(m.Value, &e); err != nil {
        return err
    }
    if _, dup := c.processed.LoadOrStore(e.ID, struct{}{}); dup {
        return nil // already processed
    }
    // simulated work
    select {
    case <-time.After(50 * time.Millisecond):
    case <-ctx.Done():
        return ctx.Err()
    }
    return nil
}

func (c *Consumer) Drain(ctx context.Context) error {
    c.draining.Store(true)
    c.fetchCancel() // stop fetching new messages

    done := make(chan struct{})
    go func() {
        c.inFlight.Wait()
        close(done)
    }()

    select {
    case <-done:
        return c.reader.Close()
    case <-ctx.Done():
        _ = c.reader.Close()
        return ctx.Err()
    }
}

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer cancel()

    c := NewConsumer([]string{"localhost:9092"}, "events", "demo")
    c.Start(ctx, 8)

    <-ctx.Done()
    log.Println("draining")

    dctx, dcancel := context.WithTimeout(context.Background(), 25*time.Second)
    defer dcancel()

    if err := c.Drain(dctx); err != nil {
        log.Printf("drain: %v", err)
    }
    log.Println("exit")
}

Walk through this carefully. Note:

• Two contexts: parentCtx (passed by main, cancelled on signal) and fetchCtx (derived, cancelled on drain to stop fetching but allow in-flight workers to complete).
• The fetcher closes msgs on exit; workers range over it.
• Workers commit after process; failures don't commit (next consumer will retry).
• Drain stops fetching, waits for workers to drain msgs, then closes the reader.
• Idempotency map ensures duplicates from at-least-once delivery are filtered.

This is real, working, drain-aware Kafka consumer code. Adapt to your topic, group, and processing logic.

Extended Section: Drain In Pulsar¶

Pulsar's Go client has a similar shape. Drain via Consumer.Close():

consumer.Close() // blocks until pending acks complete

For finer control, track in-flight as in the Kafka example. Pulsar's pattern is "ack each message after processing; Close flushes pending acks." Similar to Kafka but with different APIs.

Extended Section: Drain In NATS JetStream¶

NATS JetStream has built-in Drain:

sub.Drain() // stop receiving, process pending, close

The NATS library is the cleanest example of drain-as-first-class. Other libraries should aspire to this.

Extended Section: Drain In Kinesis¶

Kinesis is shard-based. Drain considers the shard iterator:

client.GetRecords(ctx, &kinesis.GetRecordsInput{ShardIterator: it})

On drain:

1. Stop calling GetRecords for new batches.
2. Process the last fetched batch.
3. Checkpoint via DynamoDB (if using KCL pattern).

Kinesis lacks Kafka's built-in offset commit; you implement checkpointing. Drain must checkpoint cleanly.

Extended Section: Drain In SQS¶

SQS is simpler:

sqs.ReceiveMessage(ctx, ...) // long poll
// ... process ...
sqs.DeleteMessage(ctx, ...) // ack

Drain:

1. Stop calling ReceiveMessage.
2. Process and delete current batch.
3. Done.

SQS has no offset; deletion is the only state. Drain is the most straightforward of the message systems.

Extended Section: Drain Across All Of Them¶

Common pattern:

type StreamConsumer interface {
    Fetch(ctx context.Context) (Message, error)
    Ack(ctx context.Context, m Message) error
    Close() error
}

type Drainer struct {
    c        StreamConsumer
    inFlight sync.WaitGroup
    cancel   context.CancelFunc
}

func (d *Drainer) Drain(ctx context.Context) error {
    d.cancel() // stop fetching
    done := make(chan struct{})
    go func() { d.inFlight.Wait(); close(done) }()
    select {
    case <-done:
        return d.c.Close()
    case <-ctx.Done():
        _ = d.c.Close()
        return ctx.Err()
    }
}

This abstraction works for Kafka, Pulsar, NATS, Kinesis, SQS. The specific Fetch and Ack implementations differ; the drain wrapper does not.

Extended Section: A Professional-Level Checklist¶

For any messaging-driven service, the checklist before production:

• Consumer drain stops fetch and waits for in-flight.
• Workers commit / ack only after successful processing.
• Drain calls Close after all workers exit.
• Idempotency mechanism handles duplicates.
• Transactions (if any) commit or abort cleanly on drain.
• Rebalance hooks implemented for Kafka.
• Drain budget exceeds longest reasonable transaction.
• Drain budget less than grace period.
• Drain test in CI with simulated load.
• Drain metrics emitted; alerts configured.
• Goroutine leak test in CI.
• Drain duration tracked across deploys.

Twelve items. Each takes hours of engineering. The cumulative payoff is a service that ships fearlessly.

Closing Thoughts At Professional Level¶

Drain at the professional level is the discipline of getting messaging right. Kafka, with its rebalance protocol and exactly-once semantics, is the most demanding case. The patterns extend to every other streaming system.

If you can drain a Kafka consumer cleanly with EOS, you can drain anything. The patterns from this page are reusable; the discipline transfers.

Master this, and you can build systems that handle billions of messages per day with no data loss, even across constant deploys, scale events, and rebalances. That is the engineering bar.

Welcome to the deep end. The water is fine.

Appendix A: Drain Behaviour Across Kafka Library Versions¶

A short reference. As Kafka libraries evolve, drain semantics evolve too.

sarama¶

Older versions of sarama had bugs in ConsumerGroup.Close — it could hang on shutdown. Modern versions (1.40+) handle drain properly via Cleanup callbacks.

confluent-kafka-go¶

Built on librdkafka. Drain via consumer.Close(). Cgo means goroutines can leak across the boundary; pay attention to runtime.LockOSThread usage.

kafka-go¶

Pure Go. Drain via reader.Close(). Cleaner than sarama for simple cases.

franz-go¶

Newer, supports cooperative rebalance. Drain via client.Close() and explicit lifecycle methods.

Choose based on your needs. For most new projects in 2026, franz-go is a strong default.

Appendix B: Drain Behaviour Across Pulsar Versions¶

The Apache Pulsar Go client has evolved similarly:

• Older versions had buggy Close.
• Modern versions (0.10+) drain pending acks before exit.

Always upgrade to the latest minor; drain behaviour improves.

Appendix C: A Performance Note On Drain¶

At very high throughput (100k+ msg/sec), drain may be the bottleneck of pod cycling:

• A 10s drain × 100 pods × 1 deploy/day = 1000s of "drain time" per day.
• At 100k msg/sec per pod, that is 100M messages "in drain" per day.

If drain wastes capacity (idle workers waiting), the throughput cost is real. Optimisations:

• Parallel drain across components.
• Shorter drain budgets when safe.
• Pre-emptive shedding of low-priority work.

This is the senior+ thinking at high throughput.

Appendix D: Drain Failure Modes Recap For Kafka¶

Specific failure modes when drain goes wrong on Kafka:

1. Duplicate processing. Offset not committed before disconnect.
2. Stuck rebalance. Consumer takes too long in Cleanup.
3. Lag spike. Drain interrupts mid-batch; partition stalled.
4. Coordinator timeout. Drain longer than session.timeout.ms.
5. Producer error on close. Buffered messages not flushed.
6. Transaction abort. EOS work discarded.
7. Idempotency loss. Producer ID resets; sequence number conflicts.

Each has a specific mitigation. Drain code must be aware of them.

Appendix E: A Final Long Worked Example¶

A complete service: HTTP API accepting orders, Kafka consumer processing them, Kafka producer emitting events, Postgres persisting state. Drain across all of them.

package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "errors"
    "log"
    "net/http"
    "os"
    "os/signal"
    "sync"
    "sync/atomic"
    "syscall"
    "time"

    "github.com/segmentio/kafka-go"
    _ "github.com/lib/pq"
)

type Order struct {
    ID     string  `json:"id"`
    Amount float64 `json:"amount"`
}

type App struct {
    db        *sql.DB
    consumer  *kafka.Reader
    producer  *kafka.Writer
    inFlight  sync.WaitGroup
    draining  atomic.Bool
    fetchCtx  context.Context
    fetchCancel context.CancelFunc
}

func (a *App) Start(ctx context.Context, workers int) {
    a.fetchCtx, a.fetchCancel = context.WithCancel(ctx)
    msgs := make(chan kafka.Message, 64)
    go a.fetcher(msgs)
    for i := 0; i < workers; i++ {
        a.inFlight.Add(1)
        go a.worker(ctx, i, msgs)
    }
}

func (a *App) fetcher(msgs chan<- kafka.Message) {
    defer close(msgs)
    for {
        m, err := a.consumer.FetchMessage(a.fetchCtx)
        if err != nil {
            return
        }
        select {
        case <-a.fetchCtx.Done():
            return
        case msgs <- m:
        }
    }
}

func (a *App) worker(ctx context.Context, id int, msgs <-chan kafka.Message) {
    defer a.inFlight.Done()
    for m := range msgs {
        var o Order
        if err := json.Unmarshal(m.Value, &o); err != nil {
            log.Printf("[%d] decode: %v", id, err)
            continue
        }
        if err := a.processOrder(ctx, o); err != nil {
            log.Printf("[%d] process: %v", id, err)
            continue
        }
        if err := a.consumer.CommitMessages(ctx, m); err != nil {
            log.Printf("[%d] commit: %v", id, err)
        }
    }
}

func (a *App) processOrder(ctx context.Context, o Order) error {
    tx, err := a.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer func() { _ = tx.Rollback() }()
    if _, err := tx.ExecContext(ctx,
        "INSERT INTO orders (id, amount) VALUES ($1, $2) ON CONFLICT DO NOTHING",
        o.ID, o.Amount); err != nil {
        return err
    }
    if err := tx.Commit(); err != nil {
        return err
    }
    evt, _ := json.Marshal(o)
    return a.producer.WriteMessages(ctx, kafka.Message{
        Key:   []byte(o.ID),
        Value: evt,
    })
}

func (a *App) Drain(ctx context.Context) error {
    a.draining.Store(true)
    a.fetchCancel()
    done := make(chan struct{})
    go func() { a.inFlight.Wait(); close(done) }()
    select {
    case <-done:
    case <-ctx.Done():
        return ctx.Err()
    }
    if err := a.consumer.Close(); err != nil {
        log.Printf("consumer close: %v", err)
    }
    if err := a.producer.Close(); err != nil {
        log.Printf("producer close: %v", err)
    }
    return a.db.Close()
}

func (a *App) Handler() http.Handler {
    mux := http.NewServeMux()
    mux.HandleFunc("/ready", func(w http.ResponseWriter, r *http.Request) {
        if a.draining.Load() {
            w.WriteHeader(http.StatusServiceUnavailable)
            return
        }
        w.WriteHeader(http.StatusOK)
    })
    return mux
}

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer cancel()

    db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
    if err != nil {
        log.Fatal(err)
    }
    consumer := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"localhost:9092"},
        GroupID: "orders",
        Topic:   "orders",
    })
    producer := &kafka.Writer{
        Addr:  kafka.TCP("localhost:9092"),
        Topic: "order_events",
    }

    app := &App{db: db, consumer: consumer, producer: producer}
    app.Start(ctx, 8)

    srv := &http.Server{Addr: ":8080", Handler: app.Handler()}
    go func() {
        err := srv.ListenAndServe()
        if !errors.Is(err, http.ErrServerClosed) {
            log.Printf("http: %v", err)
        }
    }()

    <-ctx.Done()
    log.Println("draining")
    start := time.Now()

    dctx, dcancel := context.WithTimeout(context.Background(), 25*time.Second)
    defer dcancel()

    app.draining.Store(true)
    time.Sleep(2 * time.Second)

    if err := srv.Shutdown(dctx); err != nil {
        log.Printf("http shutdown: %v", err)
    }
    if err := app.Drain(dctx); err != nil {
        log.Printf("drain: %v", err)
    }
    log.Printf("drained in %s", time.Since(start))
}

This is approximately 130 lines. It is a real, working, drain-aware service with Kafka, Postgres, and HTTP. Read it carefully. Identify each pattern from this page. Adapt to your stack.

Final Words¶

You have reached the end of professional-level drain. The patterns are demanding; the discipline is real. Drain at this level is not "graceful shutdown" — it is the foundation of exactly-once messaging in production.

If you can implement this code from memory, with all its subtleties, you have mastered the drain pattern in Go. There is always more to learn (specific cloud quirks, specific framework bugs), but the foundation is solid.

Go ship systems that drain cleanly. The world is full of buggy shutdown code; be the engineer who fixes it.

Appendix F: Drain In Multi-Datacenter Kafka¶

Some organisations run MirrorMaker or equivalent to replicate between Kafka clusters across datacenters. Drain interacts:

• Replicators are themselves consumers + producers.
• Their drain must commit offsets in both clusters.
• A drain that fails in the middle may leave clusters out of sync.

Best practices:

• Replicators have their own drain budgets.
• Monitoring tracks replication lag during drain.
• Lag spikes during deploys are normal; alerts have buffer.

For services running across DCs, drain considers the slower DC's grace period as the bound.

Appendix G: Drain And Schema Evolution¶

Schema evolution in messaging systems (Avro, Protobuf) interacts with drain:

• A new schema version may require both producer and consumer changes.
• During rolling deploy, old and new versions coexist.
• A consumer draining mid-rollout may have a mix of schema versions in flight.

Drain code must handle both schemas. Test with mixed-version traffic.

Appendix H: Drain And Schema Registry¶

If using a schema registry:

• Drain does not need to disconnect from the registry (it is rarely a bottleneck).
• But if registry calls are slow, drain may be delayed.

A drain that calls the registry should bound the call. Cached schemas are immune.

Appendix I: Drain In Stateful Stream Processors¶

A stateful stream processor (Kafka Streams-like) holds state in RocksDB or similar. Drain must:

1. Stop processing input.
2. Wait for state writes to complete.
3. Flush RocksDB write buffers.
4. Snapshot state for next instance.
5. Commit input offsets.
6. Disconnect.

The state snapshot can be large. Drain budget must accommodate the time to write it.

Mitigations:

• Continuous incremental snapshots (so the final snapshot is small).
• Stand-by replicas (so drain does not need a full snapshot).
• Streaming state migration (next instance reads from this one).

These are advanced patterns; relevant for high-state services.

Appendix J: Drain Cohort¶

A "drain cohort" is a group of pods drained together. Useful for batch deploys.

In Kubernetes, define a PodDisruptionBudget:

spec:
  minAvailable: 90%

The deploy never disrupts more than 10% at a time. The drain cohort is 10% of the fleet.

For large fleets (100+ pods), cohort drain is essential. Single-pod-at-a-time deploys take too long.

Appendix K: Drain Health Models¶

A pod's health during drain has nuances:

• Liveness: still alive (responding to liveness probes).
• Readiness: not ready for new traffic.
• Healthiness for the cluster: still contributing (e.g., processing in-flight).

During drain, the pod should be: alive yes, ready no, healthy yes-until-drain-complete.

Implementation:

• Liveness handler: always 200 unless catastrophic.
• Readiness handler: 503 during drain.
• Custom "cluster health" handler: 200 until in-flight is zero, then 503.

A control plane that watches all three can orchestrate sophisticated drain scenarios.

Appendix L: Drain Vs Restart Strategy¶

In Kubernetes, Recreate vs RollingUpdate deploy strategies interact with drain:

• Recreate: all old pods drain, then all new pods start. Drain happens in parallel.
• RollingUpdate: one pod drains, one starts, repeat. Drain happens serially.

For exactly-once messaging, RollingUpdate is safer. For stateless services, Recreate may be faster.

Senior engineers choose based on workload characteristics.

Appendix M: Drain And Service Mesh Retries¶

A service mesh that retries requests on 5xx may inadvertently amplify load during drain:

• Pod A drains; returns 503 for new requests.
• Mesh retries on 503 to pod B.
• Pod B is also draining; returns 503.
• Mesh retries to pod C.

If many pods drain simultaneously, this cascades.

Mitigations:

• Mesh respects Retry-After headers.
• Drain returns a non-retryable status (e.g., 503 with no retry header, or 410 Gone).
• Mesh has a retry budget per source pod.

Appendix N: Drain Across Bounded Contexts¶

In a service composed of multiple bounded contexts (sub-services within one binary):

• Each bounded context has its own drain.
• The supervisor coordinates them.
• Inter-context calls are bounded by the drain context.

This is essentially monorepo-internal microservices, drained as a unit.

Appendix O: Drain Of Lambda Functions¶

AWS Lambda functions have their own drain semantics:

• A function invocation is short-lived (max 15 minutes).
• Drain is per-invocation, not per-process.
• The runtime exposes lifecycle events: init, invoke, shutdown.

For Go on Lambda:

runtime.RegisterExtension("graceful-shutdown", func(event LifecycleEvent) {
    if event.Type == "shutdown" {
        // flush logs, close connections
    }
})

Lambda extensions get up to 2 seconds during shutdown. Drain accordingly.

Appendix P: Drain Of Knative Services¶

Knative auto-scales pods. Drain interacts with auto-scaling:

• Scale-down drain follows the standard pattern.
• Scale-to-zero is a long drain (no requests for a while, then drain).
• Cold start after scale-from-zero has no drain — it starts fresh.

For Knative, drain semantics are the same as Kubernetes. The auto-scaler is the orchestrator.

Appendix Q: Drain Beyond Messaging¶

Beyond Kafka and similar, drain applies to:

• WebSocket gateways.
• Long-poll HTTP servers.
• gRPC streaming.
• Server-sent events.
• Custom TCP servers.

The patterns are the same. The protocol-specific details differ.

Appendix R: Drain In Distributed Lock Holders¶

A service holding distributed locks (etcd, Redis Redlock) must release them on drain.

type LockedService struct {
    locks []*Lock
}

func (s *LockedService) Drain(ctx context.Context) error {
    var wg sync.WaitGroup
    for _, l := range s.locks {
        l := l
        wg.Add(1)
        go func() { defer wg.Done(); _ = l.Release(ctx) }()
    }
    done := make(chan struct{})
    go func() { wg.Wait(); close(done) }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Release in parallel; total time is one RPC roundtrip.

Appendix S: Drain Of Caching Layers¶

A write-through cache flushes dirty entries on drain.

func (c *Cache) Drain(ctx context.Context) error {
    c.mu.RLock()
    defer c.mu.RUnlock()
    for k, v := range c.dirty {
        if err := c.backend.Set(ctx, k, v); err != nil {
            return err
        }
    }
    return nil
}

Drain time is proportional to the dirty set size. For large caches, this can be slow.

Mitigations:

• Write-through (always write to backend) avoids drain flush.
• Bounded dirty set size.
• Background flush worker drains the set continuously.

Appendix T: Drain Of Search Indices¶

A service with an in-memory search index (Bleve, Tantivy) must persist the index on drain.

func (s *Search) Drain(ctx context.Context) error {
    return s.index.Persist(ctx)
}

Index persist can be slow (gigabytes). Drain budget must accommodate.

For very large indices, replicate continuously to a stand-by; drain becomes a no-op.

Appendix U: Drain Of Background Schedulers¶

A background scheduler (e.g., a job runner that polls a queue) drains by stopping the poll and waiting for in-flight jobs.

type Scheduler struct {
    stop     chan struct{}
    inFlight sync.WaitGroup
}

func (s *Scheduler) Drain(ctx context.Context) error {
    close(s.stop)
    done := make(chan struct{})
    go func() { s.inFlight.Wait(); close(done) }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

The pattern repeats. Different sources, same shape.

Appendix V: Drain Across Service Discovery¶

A service registered in service discovery (Consul, etcd) must deregister on drain.

func (s *Service) Drain(ctx context.Context) error {
    if err := s.discovery.Deregister(ctx); err != nil {
        return err
    }
    // then drain workload
    return s.workload.Drain(ctx)
}

Deregister early so new traffic stops. Then drain.

For platforms that auto-deregister on connection close (e.g., Consul via TTL), this is automatic. For those that require explicit deregister, it is a drain step.

Appendix W: Drain Of Background Tasks¶

A "background tasks" library (queue jobs, cron, etc.) drain via standard patterns.

type Tasks struct {
    tasks   chan Task
    workers sync.WaitGroup
    closed  atomic.Bool
}

func (t *Tasks) Drain(ctx context.Context) error {
    if !t.closed.CompareAndSwap(false, true) {
        return nil
    }
    close(t.tasks)
    done := make(chan struct{})
    go func() { t.workers.Wait(); close(done) }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Same pattern as worker pools, with tasks instead of jobs.

Appendix X: Drain Of CRDTs¶

A service replicating CRDTs (conflict-free replicated data types) must:

• Stop accepting writes.
• Propagate pending writes to replicas.
• Achieve eventual consistency before drain ends.

The drain must accommodate the slowest replica's catch-up. For globally-distributed CRDTs, this can be seconds.

Appendix Y: Drain In Edge Compute¶

Edge platforms (Cloudflare Workers, Fastly) have minimal drain:

• Each invocation is isolated.
• No persistent state between invocations.
• Drain is per-invocation, not process-level.

For Go compiled to WASM on the edge, drain is mostly about flushing the current request. The platforms handle cluster-level drain.

Appendix Z: Drain Of Stream Joins¶

A stream join (Kafka Streams-like join) drains by:

1. Flushing the join state.
2. Committing both source streams' offsets.
3. Persisting the join state for resumption.

Joins are expensive to drain. Snapshotting can take seconds for large state. Plan budget accordingly.

Appendix AA: Drain In Multi-Tenant Stream Processors¶

A processor handling N tenants in one binary drains by:

1. Stopping intake for all tenants.
2. Draining each tenant's pipeline (parallel where independent).
3. Committing per-tenant state.

Per-tenant timing varies. The drain budget is governed by the slowest tenant.

Appendix BB: Drain Cost Model¶

A simple cost model for drain:

• Drain time (sec) × pods drained × deploys/day = engineer-time wasted per day.
• For 100 pods, 10s drain, 1 deploy/day: 1000 sec/day = 17 min/day = 5.6 hours/year.

If your team values an engineer-hour at $200, that is $1100/year for drain. Across 50 services, $55,000/year.

Optimising drain to 5s saves half. The investment in drain pays back quickly.

Appendix CC: Drain Of Service Mesh Sidecars¶

Service mesh sidecars (Envoy, Linkerd) drain via their own primitives:

• Envoy: terminationDrainDuration.
• Linkerd: Linkerd2-Proxy graceful shutdown.

These run as separate containers in the pod. Both must drain inside the grace period.

Common pitfall: the application drains in 10s; the sidecar takes 25s; the pod takes 35s. The orchestrator's 30s grace period is exceeded.

Mitigation: align sidecar drain time with application drain time.

Appendix DD: Drain Across Cloud Providers Compared¶

For Go services, target the platform's default grace; adjust if your workload needs more.

Appendix EE: Drain As Documentation¶

A well-drained service is self-documenting. The drain code reveals:

• Which goroutines exist.
• Which resources are owned.
• What the dependencies are.
• How long shutdown should take.

Reading a service's Drain function is a fast way to understand the service. Make it readable.

Appendix FF: Drain In Open Source Projects¶

A few open-source Go projects with exemplary drain code:

• etcd — distributed key-value store; drain across raft consensus.
• caddy — HTTP server; clean drain via Server.Shutdown.
• prometheus — drains its scrape loops cleanly.
• tempo (Grafana traces) — drains trace ingestion pipeline.
• loki (Grafana logs) — drains log ingestion.

Read their drain code. Each is a master class in the patterns from this and earlier pages.

Appendix GG: A Final Walkthrough¶

Let us close with a walkthrough of an idealised production drain, second by second.

T=0. SIGTERM arrives. signal.NotifyContext's goroutine cancels the root context. main unblocks.

T=0.001. app.draining.Store(true). Readiness flips to 503.

T=2.0. time.Sleep(2 * time.Second) ends. LB has noticed; new traffic stops.

T=2.001. drainCtx created with 23s deadline (25s total minus 2s spent).

T=2.002. srv.Shutdown(drainCtx) called. Listener closes. Active handlers continue.

T=2.500. Last handler finishes. srv.Shutdown returns nil.

T=2.501. pool.Drain(drainCtx) called. Job channel closes.

T=4.200. All workers exit. pool.Drain returns nil.

T=4.201. producer.Flush(drainCtx) called.

T=4.500. Producer buffer flushed. producer.Flush returns nil.

T=4.501. producer.Close() called.

T=4.502. db.Close() called.

T=4.600. All connections closed. db.Close returns nil.

T=4.601. main logs drain complete in 4.6s and returns.

T=4.602. Go runtime exits with code 0.

T=4.700. Orchestrator notices process exited, marks pod terminated.

Total: 4.7 seconds. Well under the 30s grace period. No data loss. No 5xx. The deploy is invisible to customers.

Multiply by 100 pods across the org, four times a day: 400 of these per day. None of them paged anyone.

That is the goal. That is what professional drain looks like.

Appendix HH: Drain Is A Practice¶

Drain is a practice, like writing good tests or doing code reviews. It is not a one-time achievement. The patterns evolve. New libraries emerge. New failure modes appear.

Keep practising. Keep learning. Keep building services that drain cleanly. The cumulative effect on your team — and your career — is enormous.

Appendix II: Closing The Professional Chapter¶

This is the last technical content in this file. The remaining pages — specification, interview, tasks, find-bug, optimize — are reference, practice, and assessment. Use them to consolidate.

If you have absorbed this professional page, you are equipped to:

• Lead drain quality at an organisation.
• Handle the most demanding drain scenarios (EOS Kafka, complex pipelines).
• Audit any Go service for drain bugs.
• Mentor others in drain discipline.
• Contribute drain patterns to open source.

That is the bar. The work is yours.

Appendix JJ: A Final Long Reflection¶

Drain, as I write this, is one of the most underappreciated patterns in production Go. Every team that has built a serious Go service has hit a drain bug. Every senior engineer who has been on-call for a Go service has been paged at 2 AM because drain went wrong.

The good news: drain is teachable. The patterns are finite. The tooling exists. A team that invests in drain for a quarter reaps the benefits for years.

The hard part is the discipline. Drain is not glamorous. It does not appear in feature demos. It does not appear in performance benchmarks. It does not appear in promotion documents — unless you make it appear.

Make it appear. Talk about drain in design reviews. Write drain into your team's bar for "production-ready." Mentor newer engineers into drain awareness. Be the senior engineer who treats drain as first-class.

A team that does this is a team that ships fast and sleeps well. Be that team.

Appendix KK: Drain And The Future¶

What will change in the next 5 years?

• More language-level support for graceful shutdown (Go itself may evolve here).
• More framework-level drain (every framework will have it).
• More platform-level drain (orchestrators will get smarter).
• More observability tooling for drain.

But the patterns will remain. Stop intake, wait for in-flight, close downstream, bounded by deadline. That is the eternal recipe. Future systems will make it easier; the recipe is universal.

Master the recipe now. The tooling catches up.

Appendix LL: Last Words¶

You have completed the four-level journey: junior, middle, senior, professional. Drain is now in your toolkit.

Go build. Go ship. Go drain cleanly.

The work continues. The systems we leave behind us are better for it.

Appendix MM: A Last Practical Note¶

The most useful single thing you can do tomorrow: audit one service for drain quality.

Pick a service. Spend 30 minutes. Apply the 10-point checklist from the senior page. Note what is missing. Write a follow-up task.

That single audit, repeated weekly, lifts a service's drain quality over a quarter. Repeated org-wide, lifts the whole org.

Start tomorrow. Pick a service. Spend 30 minutes.

The rest follows.

Appendix NN: Deep Dive — Kafka Rebalance Internals¶

To drain Kafka consumers correctly, you need to understand the rebalance protocol at the broker level.

The group coordinator¶

Each consumer group has a coordinator — one of the brokers, chosen via a hash of the group ID. The coordinator:

• Tracks group membership.
• Assigns partitions to members.
• Tracks heartbeats and session timeouts.
• Initiates rebalances when membership changes.

A consumer's "drain" interacts with the coordinator in two ways:

1. When the consumer's session expires (no heartbeat), the coordinator marks it dead and rebalances.
2. When the consumer cleanly leaves the group (LeaveGroup request), the coordinator rebalances immediately.

A drain should do (2) — call LeaveGroup explicitly. Most Go libraries do this in Close().

Generation IDs¶

Each rebalance produces a new generation ID. Consumers that miss a heartbeat may have a stale generation and be rejected. Drain interacts:

• Long drain may cause heartbeats to lag.
• The consumer is then booted from the group.
• Any commits with the stale generation are rejected.

Mitigation: keep heartbeats running during drain. Most libraries do this on a background goroutine that runs until Close.

Static vs dynamic membership¶

Static membership (Kafka 2.3+) lets a consumer keep its partition assignment across restarts (up to session.timeout.ms). A draining consumer:

• Leaves cleanly via LeaveGroup.
• A replacement with the same group.instance.id reclaims the assignment.

This is useful for stateful consumers (stream processors) where the new instance can pick up from the same state.

Range vs round-robin vs sticky assignment¶

The assignment strategy affects drain:

• Range: contiguous partitions per consumer. Drain triggers contiguous reassignment.
• Round-robin: spread across consumers. Drain triggers wider reassignment.
• Sticky: minimise movement on rebalance. Drain only moves the leaving consumer's partitions.

For drain, sticky is best — least disruption. Use it when available.

Cooperative rebalance protocol¶

The cooperative rebalance protocol (CooperativeStickyAssignor in Java) breaks rebalance into two phases:

1. Revoke phase: each consumer revokes only the partitions it is about to lose. Others continue.
2. Assign phase: new assignments are applied.

The Go ecosystem is catching up: franz-go supports cooperative; older libraries default to eager.

Cooperative changes drain semantics:

• Drain may revoke only some partitions, not all.
• The consumer must drain those partitions specifically.
• Other partitions continue uninterrupted.

This requires per-partition drain logic.

Appendix OO: Per-Partition Drain Implementation¶

For cooperative rebalance, drain is per-partition.

type PartitionRunner struct {
    partition int32
    in        chan kafka.Message
    wg        sync.WaitGroup
    committed atomic.Int64
}

type Consumer struct {
    mu         sync.Mutex
    runners    map[int32]*PartitionRunner
    commit     func(p int32, offset int64) error
}

func (c *Consumer) OnPartitionsRevoked(parts []int32) error {
    for _, p := range parts {
        c.mu.Lock()
        r := c.runners[p]
        delete(c.runners, p)
        c.mu.Unlock()
        if r == nil {
            continue
        }
        close(r.in)
        r.wg.Wait()
        if err := c.commit(p, r.committed.Load()); err != nil {
            log.Printf("commit %d: %v", p, err)
        }
    }
    return nil
}

func (c *Consumer) OnPartitionsAssigned(parts []int32) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    for _, p := range parts {
        r := &PartitionRunner{
            partition: p,
            in:        make(chan kafka.Message, 16),
        }
        c.runners[p] = r
        r.wg.Add(1)
        go c.runPartition(r)
    }
    return nil
}

func (c *Consumer) runPartition(r *PartitionRunner) {
    defer r.wg.Done()
    for msg := range r.in {
        _ = c.process(msg)
        r.committed.Store(msg.Offset + 1)
    }
}

The per-partition runner has its own goroutine, channel, and offset state. Revocation closes the channel; the runner drains its buffer and exits; the offset is committed.

This is more code than the simple consumer, but it handles cooperative rebalance correctly.

Appendix PP: Drain Order Within A Kafka Consumer¶

Inside a single Kafka consumer, drain order matters:

1. Stop fetcher. No new messages enter the system.
2. Drain per-partition queues. Each partition's in-flight work completes.
3. Commit offsets per partition. The highest processed offset is recorded.
4. Heartbeat one last time. The coordinator sees us as alive until the last moment.
5. LeaveGroup. Tell the coordinator we are leaving cleanly.
6. Close connection. TCP-level shutdown.

Skipping any step risks duplicates or lag.

Appendix QQ: Drain Error Recovery¶

Drain can fail. Strategies:

• Idempotent re-drain. Calling drain again after a failure tries again.
• Partial-state recovery. Note which components drained; skip them on retry.
• Force-cleanup. If clean drain fails, close hard.

func (s *Service) DrainWithRetry(ctx context.Context, retries int) error {
    for i := 0; i < retries; i++ {
        if err := s.Drain(ctx); err == nil {
            return nil
        }
    }
    s.ForceClose()
    return errors.New("drain failed; force-closed")
}

For most services, one drain attempt is enough. Retries are useful for transient errors (network blip during commit).

Appendix RR: Drain And End-to-End Latency¶

Drain affects end-to-end latency:

• During drain, in-flight requests have an upper bound (the drain deadline).
• Slow requests may be force-cancelled.
• The P99 of "drain-affected" requests is higher.

Track P99 latency by deploy time. A spike during deploys indicates drain pressure.

Appendix SS: Drain And SLA Compliance¶

Service-Level Agreements often specify uptime (99.9%, 99.99%). Drain affects uptime:

• A drain that returns 5xx counts as downtime.
• A drain that drops requests counts as downtime.
• Frequent deploys with drain issues accumulate downtime.

Compute the impact: drain 5xx rate × deploy frequency × pod-time = uptime loss.

For a 99.95% SLA, this matters. Drain quality is part of compliance.

Appendix TT: Drain And Cost Optimisation¶

Drain consumes capacity:

• A drain pod is not serving full traffic.
• The replacement is starting up.
• For minutes, the cluster has reduced capacity.

For cost-optimised deployments (running near capacity), drain time directly affects ability to deploy. If drain is 30s and you cannot tolerate any capacity loss, you cannot deploy.

Mitigation: keep drain short; or over-provision modestly during deploys.

Appendix UU: Drain And Auto-Scaling¶

Auto-scaling interacts with drain:

• Scale-down drains existing pods.
• Scale-up replaces drained capacity quickly.
• Rapid scale changes can amplify drain pressure.

For latency-sensitive services, scale changes should be slow. For throughput services, fast scale changes are fine.

Drain budget must accommodate scale-down events. If you scale down 50% of pods at once, drain handles 2x normal load briefly.

Appendix VV: Drain Comparison Across Languages — Long Version¶

Go's drain story is solid. Erlang is even cleaner. Java/.NET are more verbose but capable.

When evaluating a polyglot stack, ensure each language is drainable. A single weak link is a single point of failure.

Appendix WW: Drain And Performance Testing¶

A performance test that excludes drain misses key data. Include drain in the test:

# Load for 60 seconds.
vegeta attack -duration 60s -rate 1000 < targets > results
sleep 30
kill -TERM $SERVER_PID
wait $SERVER_PID
# Verify exit code 0
# Verify zero 5xx in last 10 seconds of result file

The "last 10 seconds" is the drain window. Anomalies there are drain bugs.

Appendix XX: Drain And Chaos Engineering¶

Inject random SIGTERM during sustained load:

for i in {1..100}; do
    ./service &
    PID=$!
    SLEEP_TIME=$(( (RANDOM % 30) + 5 ))
    sleep $SLEEP_TIME
    kill -TERM $PID
    wait $PID
    if [ $? -ne 0 ]; then
        echo "iter $i failed"
        exit 1
    fi
done

100 iterations of random drain. Any failure is a bug.

This is a poor man's chaos test. Production teams use tools like ChaosMesh or Litmus for more sophisticated chaos.

Appendix YY: Drain Of Background Schedulers — Cron-like¶

A cron-like scheduler in Go drains by:

1. Stopping the tick loop.
2. Letting any running job complete (or force-cancel on deadline).
3. Persisting any state (last-run-time).

type Cron struct {
    jobs    []*Job
    stop    chan struct{}
    running sync.WaitGroup
}

func (c *Cron) Run(ctx context.Context) {
    t := time.NewTicker(time.Minute)
    defer t.Stop()
    for {
        select {
        case <-c.stop:
            return
        case <-ctx.Done():
            return
        case now := <-t.C:
            for _, j := range c.jobs {
                if j.Due(now) {
                    c.running.Add(1)
                    go func(j *Job) { defer c.running.Done(); j.Run(ctx) }(j)
                }
            }
        }
    }
}

func (c *Cron) Drain(ctx context.Context) error {
    close(c.stop)
    done := make(chan struct{})
    go func() { c.running.Wait(); close(done) }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Cron drain is straightforward. The complexity is in long-running jobs that may exceed drain budget.

Appendix ZZ: Drain Of Cron Jobs With Long Runners¶

If a cron job takes 5 minutes and drain budget is 25 seconds, you have a problem. Options:

1. Persist the long job's state; resume on next pod.
2. Run long jobs in a separate pod with longer grace.
3. Skip the long job during drain.

Choose based on the job's criticality. A daily report can be skipped; a billing run cannot.

Appendix AAA: Drain Library Survey¶

Open-source Go libraries for drain helpers:

• golang.org/x/sync/errgroup — error group.
• go.uber.org/fx — dependency injection with lifecycle hooks.
• github.com/oklog/run — concurrent actors.
• github.com/heptiolabs/healthcheck — health endpoints.

None of them are full drain frameworks. They provide pieces. Compose them.

For org-wide drain, build your own framework on top of these.

Appendix BBB: Drain Framework Anti-Patterns¶

Some framework anti-patterns:

• Framework that requires import _ "github.com/.../init". Init runs at import time; no control.
• Framework that catches SIGTERM automatically. Then your main cannot control drain.
• Framework that prints to stdout. Pollutes logs.
• Framework that doesn't expose drain duration metric. Cannot tune.
• Framework that doesn't support custom drain logic. Cannot adapt.

A good drain framework is opt-in, configurable, observable. Evaluate before adopting.

Appendix CCC: Building A Production Drain Library¶

The pseudocode for a production drain library:

package drain

import (
    "context"
    "sync"
    "time"
)

type Drainable interface {
    Name() string
    Drain(ctx context.Context) error
}

type Manager struct {
    mu       sync.Mutex
    items    []Drainable
    timeout  time.Duration
    onStart  []func()
    onEnd    []func(time.Duration, error)
}

func New(timeout time.Duration) *Manager {
    return &Manager{timeout: timeout}
}

func (m *Manager) Register(d Drainable) {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.items = append(m.items, d)
}

func (m *Manager) OnStart(fn func()) {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.onStart = append(m.onStart, fn)
}

func (m *Manager) OnEnd(fn func(time.Duration, error)) {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.onEnd = append(m.onEnd, fn)
}

func (m *Manager) Drain() error {
    for _, fn := range m.onStart {
        fn()
    }
    start := time.Now()
    ctx, cancel := context.WithTimeout(context.Background(), m.timeout)
    defer cancel()

    var firstErr error
    for i := len(m.items) - 1; i >= 0; i-- {
        d := m.items[i]
        if err := d.Drain(ctx); err != nil && firstErr == nil {
            firstErr = err
        }
    }

    elapsed := time.Since(start)
    for _, fn := range m.onEnd {
        fn(elapsed, firstErr)
    }
    return firstErr
}