Imported from schretzi/schretzi-skills (skills/golang-kafka/SKILL.md). Install upstream with npx skills add schretzi/schretzi-skills --skill golang-kafka. Copyright stays with the author (MIT).

Persona: You are a Go streaming engineer. You treat the schema as the API contract, assume every consumer will be restarted mid-batch, and design for at-least-once delivery until someone proves exactly-once is worth its cost.

Modes:

• Build — new producer or consumer: settle the schema and topic design first (see Schema first), then wire the client.
• Extend — adding a field, a topic, or a consumer to an existing pipeline: check compatibility mode before touching the schema.
• Debug — lag, rebalance storms, duplicate or missing messages: start at Common Mistakes, which is ordered by how often each one is the actual cause.

Dependencies:

• go get github.com/twmb/franz-go — client (recommended)
• go get github.com/twmb/franz-go/pkg/sr — schema registry + Confluent wire format
• go get github.com/linkedin/goavro/v2 — Avro codec (see Avro library warning)

Kafka streaming in Go

Defaults for this stack

Avro library choice — read this first

Do not use github.com/hamba/avro/v2. It is the most-recommended Avro library in blog posts and it is now archived (no commits since 2026-01-18), with three reachable vulnerabilities that will never be fixed:

govulncheck reports Fixed in: N/A for all three. Any project using it fails a govulncheck gate — including make pipeline in projects scaffolded by schretzi/schretzi-skills@golang-project-scaffolding.

Two clean alternatives, both verified with govulncheck:

Pick goavro unless you are migrating an existing hamba codebase, where the fork is a one-line import change.

Re-check before committing to either:

go list -m -u all | grep avro
govulncheck ./...

Schema first, always

The schema is the contract between teams that deploy independently. Decide it before writing client code.

1. Write the .avsc and commit it to the repo. It is source, not a generated artifact.
2. Register it against the subject, and let the registry reject incompatible changes — that check is the point of running a registry.
3. Set the compatibility mode explicitly on the subject. The cluster default is often BACKWARD, but never rely on it being what you assume.

What is safe to change

A rename is three deploys, not one. Plan for it or avoid it.

The Confluent wire format

Every Avro message on Kafka carries a five-byte header. Getting this wrong is the single most common integration failure between Go and JVM services.

┌────────┬──────────────────┬──────────────────────────┐
│ 0x00   │ schema ID        │ Avro binary payload      │
│ 1 byte │ 4 bytes, big-end │ n bytes                  │
└────────┴──────────────────┴──────────────────────────┘

The payload is binary Avro without an embedded schema — not Avro OCF (Object Container File). A consumer reads the ID, fetches that exact writer schema from the registry, and resolves it against its own reader schema.

Never hand-roll this. franz-go/pkg/sr implements it:

import "github.com/twmb/franz-go/pkg/sr"

rcl, err := sr.NewClient(sr.URLs("http://schema-registry:8081"))
if err != nil {
    return fmt.Errorf("schema registry client: %w", err)
}

ss, err := rcl.CreateSchema(ctx, "orders-value", sr.Schema{
    Schema: string(schemaText),
    Type:   sr.TypeAvro,
})
if err != nil {
    return fmt.Errorf("registering orders-value: %w", err)
}

var serde sr.Serde
serde.Register(ss.ID, Order{},
    sr.EncodeFn(func(v any) ([]byte, error) {
        return codec.BinaryFromNative(nil, toNative(v.(Order)))
    }),
    sr.DecodeFn(func(b []byte, v any) error {
        native, _, err := codec.NativeFromBinary(b)
        if err != nil {
            return err
        }
        return fromNative(native, v.(*Order))
    }),
)

serde.Encode prepends the header; serde.Decode strips it and dispatches on the ID. serde.DecodeID(b) returns the ID and remaining bytes when you need to route on schema before decoding.

Consumers must handle IDs they have never seen — a producer may deploy a new schema version first. Fetch unknown IDs from the registry at runtime and cache them; do not assume the set of IDs is known at startup.

Client: franz-go

→ See references/franz-go.md for the full producer, consumer-group and transaction API.

Producer defaults that matter

cl, err := kgo.NewClient(
    kgo.SeedBrokers(brokers...),
    kgo.RequiredAcks(kgo.AllISRAcks()),   // durability: wait for all in-sync replicas
    kgo.ProducerBatchCompression(kgo.ZstdCompression(), kgo.SnappyCompression()),
    kgo.ProducerLinger(10*time.Millisecond), // batch: throughput for a little latency
    kgo.MaxBufferedRecords(10_000),          // bound memory; back-pressures Produce
)

RequiredAcks(AllISRAcks()) is the default franz-go uses, and it is the right one — LeaderAck() silently loses data when a leader fails before replication. franz-go enables idempotent production by default, so retries do not duplicate.

Produce asynchronously with a promise in hot paths; ProduceSync per message serialises the batch away and collapses throughput:

cl.Produce(ctx, rec, func(r *kgo.Record, err error) {
    if err != nil {
        // The record is gone. Log it, count it, or route to a DLQ -
        // there is no automatic retry left at this point.
        slog.Error("produce failed", "topic", r.Topic, "key", string(r.Key), "err", err)
    }
})

Consumer group defaults

cl, err := kgo.NewClient(
    kgo.SeedBrokers(brokers...),
    kgo.ConsumerGroup("order-projector"),
    kgo.ConsumeTopics("orders"),
    kgo.Balancers(kgo.CooperativeStickyBalancer()), // no stop-the-world rebalances
    kgo.DisableAutoCommit(),                        // commit after work, not before
    kgo.OnPartitionsRevoked(func(ctx context.Context, cl *kgo.Client, _ map[string][]int32) {
        // Last chance to commit what is done. Blocking here is correct.
        if err := cl.CommitUncommittedOffsets(ctx); err != nil {
            slog.Error("commit on revoke failed", "err", err)
        }
    }),
)

CooperativeStickyBalancer should be your default. The eager balancers (Range, RoundRobin, Sticky) revoke every partition from every consumer on any membership change — a deploy of a 20-pod consumer becomes 20 full stop-the-world rebalances.

Changing balancers is itself a breaking change: cooperative and eager members cannot coexist in one group. Migrating requires a full group stop, or a two-phase rollout via a group with both balancers listed.

The poll loop

for {
    fetches := cl.PollRecords(ctx, 500)
    if fetches.IsClientClosed() {
        return nil
    }
    // Fetch errors are per-partition and mostly retryable; they are NOT
    // returned by PollRecords itself.
    fetches.EachError(func(t string, p int32, err error) {
        slog.Error("fetch", "topic", t, "partition", p, "err", err)
    })

    fetches.EachRecord(func(r *kgo.Record) {
        if err := handle(ctx, r); err != nil {
            // Decide: retry, drop, or DLQ. Do not just log and continue -
            // that is silent data loss wearing a logging costume.
        }
    })

    // Commit only what was actually processed.
    if err := cl.CommitRecords(ctx, fetches.Records()...); err != nil {
        slog.Error("commit", "err", err)
    }
}

PollRecords(ctx, n) bounds the batch; PollFetches does not, and a consumer that falls behind will hand you an unbounded slice.

Delivery semantics

Be explicit about which one you are buying, in a comment, at the poll loop.

Exactly-once does not extend past Kafka. A transaction covering a produce and a database write is not atomic — the database is not a transaction participant. For that, use the transactional outbox pattern: write the event to an outbox table in the same DB transaction as the state change, and relay it to Kafka separately.

Make consumers idempotent and at-least-once is almost always enough:

• Upserts keyed on the event's business key, not an auto-increment ID.
• A processed-message table keyed on (topic, partition, offset).
• Naturally idempotent operations ("set status to SHIPPED", not "increment").

Keys, partitions and ordering

Kafka orders messages within a partition, not within a topic. Everything else follows from that.

• The key determines the partition (hash(key) % partitions). Same key → same partition → ordered.
• Key by the entity whose ordering matters: order_id, customer_id. A null key round-robins and gives you no ordering at all.
• Partition count is effectively permanent. Increasing it rehashes keys, so a key's history splits across two partitions and ordering breaks for in-flight entities. Over-provision at creation.
• Watch for hot partitions. One key that dominates traffic (a bulk tenant, a test account) serialises onto one consumer no matter how many pods you run.
• Use log compaction for topics that are a changelog of current state — it keeps the latest value per key forever. Use time retention for event streams.

Dead-letter queues

A poison message — one that will never decode — blocks its partition forever if you retry it indefinitely. Bound the retries, then move it aside.

if err := handle(ctx, r); err != nil {
    if attempts(r) >= maxAttempts || errors.Is(err, errPermanent) {
        dlq := &kgo.Record{
            Topic: r.Topic + ".dlq",
            Key:   r.Key,
            Value: r.Value,             // the ORIGINAL bytes, undecoded
            Headers: append(r.Headers,
                kgo.RecordHeader{Key: "dlq-error", Value: []byte(err.Error())},
                kgo.RecordHeader{Key: "dlq-topic", Value: []byte(r.Topic)},
                kgo.RecordHeader{Key: "dlq-partition", Value: fmt.Appendf(nil, "%d", r.Partition)},
                kgo.RecordHeader{Key: "dlq-offset", Value: fmt.Appendf(nil, "%d", r.Offset)},
            ),
        }
        if err := cl.ProduceSync(ctx, dlq).FirstErr(); err != nil {
            return err // could not DLQ - do NOT commit, or the message vanishes
        }
    }
}

Send the original bytes, not a re-encoded value: if the failure was a decode failure, re-encoding is impossible, and the raw bytes are the evidence. Record enough headers to replay it. A DLQ nobody reads is a data-loss pipeline — alert on its depth.

Graceful shutdown

Kafka consumers are the case where a sloppy shutdown costs you duplicates on every single deploy.

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()

// 1. Stop polling; the poll loop returns when ctx is cancelled.
// 2. Finish the in-flight batch - do not abandon it.
// 3. Commit those offsets.
// 4. Close the client, which leaves the group cleanly and triggers one
//    rebalance now rather than a session-timeout stall later.
defer func() {
    if err := cl.CommitUncommittedOffsets(context.WithoutCancel(ctx)); err != nil {
        slog.Error("final commit", "err", err)
    }
    cl.Close()
}()

Commit with context.WithoutCancel(ctx) — the context that just got cancelled cannot carry the commit that cancellation makes necessary.

→ See schretzi/schretzi-skills@golang-context for WithoutCancel and propagation rules.

Client comparison

Default to franz-go. kafka-go has more stars and a gentler API, but if you are using Avro with a registry — the default here — franz-go already ships the wire format and you would otherwise be writing it yourself.

Testing

Do not mock the client. Kafka's failure modes live in rebalancing and offset handling, which a mock will not reproduce.

• Unit: test handle(ctx, record) directly — it takes bytes and returns an error, no broker needed.
• Round-trip the schema: encode and decode every schema in a test, and assert compatibility against the registered version with goavro's codec or the registry's /compatibility endpoint. Catches the "added a field without a default" class before it reaches staging.
• Integration: testcontainers-go with a real broker plus registry. A single-node cluster catches wire format and commit bugs; only a multi-node one catches replication behaviour.

→ See schretzi/schretzi-skills@golang-testing for table-driven tests and integration test structure.

Observability

Consumer lag is the metric that matters. Everything else is context for it.

• Lag per partition, not just per group — one stuck partition hides in a group average.
• Commit rate, rebalance count, and time spent in rebalance. A rising rebalance count is the leading indicator of a sick group.
• Produce error rate by topic, and DLQ depth.
• kgo.WithHooks exposes broker-level timings for metrics without wrapping every call.

→ See schretzi/schretzi-skills@golang-observability for metric naming and OpenTelemetry wiring.

Common Mistakes

Ordered by how often each is the actual cause.

Cross-References

• → schretzi/schretzi-skills@golang-context — cancellation, WithoutCancel for shutdown commits
• → schretzi/schretzi-skills@golang-concurrency — worker pools for parallel partition processing
• → schretzi/schretzi-skills@golang-error-handling — classifying permanent vs retryable failures
• → schretzi/schretzi-skills@golang-observability — lag metrics and tracing across a pipeline
• → schretzi/schretzi-skills@golang-testing — integration tests with testcontainers
• → schretzi/schretzi-skills@golang-security — SASL/TLS to brokers, and secret handling for registry auth