Kafka in production in Go: kafka-go, DLQ, Schema Registry, tuning, security

← Back to the section

If you already know what a topic, a partition, and a consumer group are, this article is about the next step: how to connect Kafka to a Go application and what you need to configure so that it runs reliably in production.

kafka-go: Writer and Reader

The segmentio/kafka-go library is the most common pure-Go Kafka client, with no cgo. To send messages you use a Writer; to receive them you use a Reader and a consumer loop.

// Send
type OrderEventPublisher struct {
	writer *kafka.Writer
}

func (p *OrderEventPublisher) Publish(ctx context.Context, event OrderEvent) error {
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}
	return p.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(event.OrderID),
		Value: payload,
	})
}

// Receive
reader := kafka.NewReader(kafka.ReaderConfig{
	Brokers: []string{"localhost:9092"},
	Topic:   "orders",
	GroupID: "billing-service",
})

for {
	msg, err := reader.ReadMessage(ctx)
	if err != nil {
		break
	}
	var event OrderEvent
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		continue
	}
	// processing
}

The consumer loop runs in a single goroutine. To process partitions in parallel, you start several goroutines — each with its own Reader and the same GroupID: Kafka distributes the partitions among them.

Commit offset: when Kafka "knows" a message has been processed

Kafka tracks how far a consumer has progressed through the messages via an offset commit. In kafka-go the moment of the commit is determined by how you read:

ReadMessage — commits right when the message is read, before processing. Simple, but a crash between reading and processing loses the message.
CommitInterval > 0 — commits in batches in the background at the given interval. Faster, but on a failure some messages arrive again.
FetchMessage + CommitMessages — manual commit. Needed when it matters to commit the offset only after a successful write to the database.

reader := kafka.NewReader(kafka.ReaderConfig{
	Brokers:        []string{"localhost:9092"},
	Topic:          "orders",
	GroupID:        "billing-service",
	CommitInterval: 0, // 0 — synchronous commit; >0 — batched in the background
})

for {
	msg, err := reader.FetchMessage(ctx)
	if err != nil {
		break
	}
	if err := handle(msg); err != nil {
		continue
	}
	reader.CommitMessages(ctx, msg) // only after successful processing
}

Message headers

Every Kafka message can carry headers — (key, value) pairs in bytes, separate from the payload. This is where you put technical metadata that does not belong in the business data:

X-Correlation-ID / traceparent — for distributed tracing.
X-Event-Version — the version of the event format.
X-Source-Service — where the message came from.

// When sending
writer.WriteMessages(ctx, kafka.Message{
	Key:   []byte(key),
	Value: payload,
	Headers: []kafka.Header{
		{Key: "X-Correlation-ID", Value: []byte(correlationID)},
	},
})

// When receiving
var correlationID string
for _, h := range msg.Headers {
	if h.Key == "X-Correlation-ID" {
		correlationID = string(h.Value)
	}
}
logger := slog.With("correlationId", correlationID)
// processing

Business data (orderId, amount, status) goes in the payload. Everything technical goes in the headers.

Dead Letter Queue: what to do with problematic messages

If the handler returns an error and the offset is not committed, the consumer keeps coming back to the same message over and over — the offset does not move and the consumer stalls. In production this means the whole group grinds to a halt.

The solution is a Dead Letter Queue (DLQ): after several failed attempts the message is moved to a separate topic (usually with a .DLT suffix), the offset is committed, and the main consumer keeps working.

The simple way: in-place retries + DLQ

There is no ready-made machinery in kafka-go — the DLQ is assembled by hand from a Reader, a Writer, and a retry loop:

func consumeWithDLQ(ctx context.Context, r *kafka.Reader, dlq *kafka.Writer) {
	for {
		msg, err := r.FetchMessage(ctx)
		if err != nil {
			return
		}
		if err := processWithRetry(ctx, msg, 3, time.Second); err != nil {
			dlq.WriteMessages(ctx, kafka.Message{
				Key:   msg.Key,
				Value: msg.Value,
				Headers: append(msg.Headers,
					kafka.Header{Key: "x-exception-message", Value: []byte(err.Error())},
					kafka.Header{Key: "x-original-topic", Value: []byte(msg.Topic)},
				),
			})
		}
		r.CommitMessages(ctx, msg)
	}
}

func processWithRetry(ctx context.Context, msg kafka.Message, attempts int, delay time.Duration) error {
	var err error
	for i := 0; i < attempts; i++ {
		if err = process(ctx, msg); err == nil {
			return nil
		}
		time.Sleep(delay)
	}
	return err
}

Here: 3 attempts with a 1-second interval, and after the third failure the message goes to orders.DLT (the topic the dlq Writer is configured for). The DLT message gets headers with the error text and the name of the original topic.

Retry topics with increasing delays

In-place retries block the partition for attempts × delay. To avoid blocking, messages are moved into separate retry topics — orders-retry-0, orders-retry-1, orders-retry-2 — with increasing delays, and after the last one into orders-dlt. The retry-topic consumer waits out the delay based on the message's write time:

// Retry-topic consumer: wait out the delay before a new attempt
msg, err := retryReader.FetchMessage(ctx)
if err != nil {
	return
}
if wait := time.Until(msg.Time.Add(30 * time.Second)); wait > 0 {
	time.Sleep(wait)
}
if err := process(ctx, msg); err != nil {
	nextWriter.WriteMessages(ctx, kafka.Message{Key: msg.Key, Value: msg.Value, Headers: msg.Headers})
}
retryReader.CommitMessages(ctx, msg)

Messages "cool off" in the retry topics without blocking the main flow.

When to use which

Temporary problem (database down, external service not responding) → retry topics with an increasing delay.
Permanent problem (invalid message, wrong format) → no retries, straight to the DLQ.
Need to distinguish error types → checks via errors.Is / errors.As: "invalid" errors go straight to the DLQ, everything else gets retried.

Schema Registry: how not to break your neighbors when the schema changes

When the producer and the consumer are different services from different teams, any change to the event structure can potentially break the consumer. Schema Registry solves this with a centralized schema store that checks compatibility.

How it works:

The producer registers the schema in Schema Registry and gets a numeric identifier (schema_id).
Into each message it writes 4 bytes with that identifier, followed by the compressed payload.
The consumer reads the schema_id, downloads the schema from Schema Registry once (then caches it), and deserializes the payload.

Avro, Protobuf, or JSON Schema

The standard in the Kafka ecosystem is Avro: a compact binary format with good support for schema evolution; in Go the types are generated with hamba/avro, and Schema Registry is accessed via the riferrei/srclient client. Protobuf is chosen by teams with gRPC infrastructure. JSON Schema is human-readable but takes several times more space — it is good for debugging.

Compatibility modes

Schema Registry checks whether a new version of the schema will break consumers that are already running:

BACKWARD (the default) — a new consumer can read old messages. You can remove optional fields and add optional ones with a default value. You cannot add required fields.
FORWARD — an old consumer can read new messages. The mirror image of BACKWARD.
FULL — both modes at once. The strictest.
NONE — no checks. Only for special cases.

Rule of thumb: BACKWARD for topics where there are many consumers and a single producer — that is the typical event bus.

Consumer lag: why the consumer falls behind

Lag is the difference between the last message in a partition and the one the consumer has reached. If lag is growing, the consumer cannot keep up with the producer. This is the main health metric for a consumer in production.

Typical causes:

Slow processing — each message makes a synchronous call to a database or an external service. Fix: parallelize the processing across goroutines, switch to batch requests.
Too few partitions per group — start more goroutines with Readers or add service instances (but no more than the number of partitions).
SessionTimeout expires — if the process freezes and stops sending heartbeats for longer than SessionTimeout (30 seconds by default in kafka-go), Kafka considers the consumer dead and triggers a partition rebalance.
GC pauses or CPU throttling in a container — the process froze for tens of seconds and lag grew by thousands of messages.

Check the lag manually:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group billing-service --describe
# TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders  0          124530          124530          0
# orders  1          124450          124530          80   ← falling behind

Metric to monitor: kafka_consumergroup_lag. Alert: lag is above N and has been growing for N minutes in a row.

Performance tuning

Kafka works well out of the box, but there are a few parameters worth knowing.

Producer (`Writer`)

Parameter	Default value	When to change
`BatchTimeout`	1 s	Lower to 5–20 ms if latency matters: the Writer waits until a batch fills up and sends it as a whole.
`BatchSize` / `BatchBytes`	100 messages / 1 MB	Raise under high write volume — more throughput thanks to larger batches.
`Compression`	none	Enable `kafka.Zstd` or `kafka.Lz4`. On JSON messages it saves 3–5x on traffic and disk space.

Consumer (`Reader`)

Parameter	Default value	When to change
`MinBytes`	1 byte	Raise to 50–500 KB. The consumer waits for data to accumulate — less load on the broker.
`MaxBytes`	1 MB	Raise if the messages are large or the stream is heavy.
`QueueCapacity`	100	Lower if each message requires long processing — a smaller buffer between fetching and processing.

The best compression algorithm for most cases is zstd (Kafka 2.1+): it compresses almost as well as gzip and runs almost as fast as lz4.

Security: SASL, SSL, ACL

By default Kafka listens on an open port with no authentication — you never leave it that way in production. Three layers of protection:

SSL/TLS — connection encryption. Certificates on the brokers and clients.
SASL — authentication. SASL/PLAIN — username and password (only over SSL). SASL/SCRAM-SHA-256 — safer: challenge-response, the password is not sent over the network. SASL/OAUTHBEARER — OAuth2 tokens, for integration with an identity provider.
ACL — authorization: who is allowed to read from and write to which topic. Managed via kafka-acls.sh.

Typical setup: one SASL user per service, with ACLs restricting its access strictly to the topics it needs.

mechanism, err := scram.Mechanism(scram.SHA256, "billing-service", os.Getenv("KAFKA_PASSWORD"))
if err != nil {
	log.Fatal(err)
}

dialer := &kafka.Dialer{
	SASLMechanism: mechanism,
	TLS:           &tls.Config{},
}

reader := kafka.NewReader(kafka.ReaderConfig{
	Brokers: []string{"broker:9093"},
	Topic:   "orders",
	GroupID: "billing-service",
	Dialer:  dialer,
})

KRaft: Kafka without ZooKeeper

Kafka used to require a separate ZooKeeper cluster to store metadata: which topics exist, who is the partition leader, which ACLs are in effect. ZooKeeper is a separate system that has to be deployed, monitored, and fixed on its own.

As of version 3.3 (2022) there is KRaft — a native Raft-based protocol running inside the brokers themselves. ZooKeeper is no longer needed. In Kafka 4.x ZooKeeper support has been removed entirely — all new clusters run on KRaft only.

For a developer this is almost invisible: bootstrap-servers and client behavior have not changed. The only thing that changes is how the ops team deploys and maintains the cluster.

In short

The Writer sends, the Reader receives. Parallelism — several goroutines with Readers in the same group.
The moment of the offset commit: ReadMessage — right on reading, CommitInterval — batched in the background, FetchMessage + CommitMessages — manually after processing.
Message headers are the place for technical metadata (trace-id, schema version). Business data goes in the payload.
The DLQ saves you from a consumer hanging on a problematic message. There is no ready-made machinery in kafka-go — retry topics and the DLQ are assembled by hand from a Reader and a Writer.
Schema Registry stores schemas and checks compatibility on changes. BACKWARD mode — a new consumer reads old messages.
Consumer lag is the main health metric for a consumer. Growing → the consumer cannot keep up.
Writer tuning: BatchTimeout, BatchBytes, Compression: kafka.Zstd. Reader tuning: MinBytes, QueueCapacity.
Security: SSL for encryption, SASL for authentication, ACL for per-topic authorization.
KRaft: as of Kafka 3.3+ ZooKeeper is not needed; in Kafka 4.x ZooKeeper support has been removed entirely.