Master Kafka Consumer Best Practices with Sarama for Faster, Safer Go Apps

Introduction

Kafka is a popular distributed, scalable, high‑performance streaming platform. When consuming data from Kafka, several best‑practice techniques can ensure processing efficiency and reliability. This article introduces these practices and demonstrates them using the Sarama Go client.

Key Kafka Consumer Best Practices

Choose the appropriate commit strategy : Kafka offers automatic and manual offset commits. Automatic commits are easy but may cause data loss or duplication; manual commits give finer control to guarantee at‑least‑once or exactly‑once processing.

Minimize the number of fetches : Reading messages in large batches greatly improves throughput. Adjust parameters such as fetch.min.bytes and fetch.max.wait.ms.

Use consumer groups whenever possible : Consumer groups enable parallel consumption and horizontal scaling while ensuring each partition is processed by only one consumer at a time.

Adjust consumer buffer sizes : Tuning receive.buffer.bytes and max.partition.fetch.bytes to match expected message size and available memory can improve consumer performance.

Handle rebalance events efficiently : Properly processing rebalances prevents pauses in consumption when consumers join or leave a group.

Monitor consumers : Use Kafka consumer metrics to identify bottlenecks and adjust configurations.

Choosing the Appropriate Commit Strategy

1. Automatic Commit

Sarama’s ConsumerGroup automatically commits offsets by default, allowing the consumer to resume from the last committed position after a restart or failure.

config := sarama.NewConfig()
config.Version = sarama.V2_0_0_0
config.Consumer.Offsets.AutoCommit.Enable = true
config.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second

ConsumerGroup, err := sarama.NewConsumerGroup(brokers, groupID, config)
if err != nil {
    log.Panicf("创建消费者组客户端时出错: %v", err)
}
Consumer := Consumer{}
ctx := context.Background()
for {
    err := ConsumerGroup.Consume(ctx, []string{topic}, Consumer)
    if err != nil {
        log.Panicf("来自消费者的错误: %v", err)
    }
}

The config.Consumer.Offsets.AutoCommit.Interval setting shows that offsets are committed every second.

2. Manual Commit

Manual commits give precise control over when offsets are recorded. The example below disables automatic commits and marks messages manually.

config := sarama.NewConfig()
config.Version = sarama.V2_0_0_0
config.Consumer.Offsets.AutoCommit.Enable = false

consumerGroup, err := sarama.NewConsumerGroup(brokers, groupID, config)
if err != nil {
    log.Panicf("创建消费者组客户端时出错: %v", err)
}
Consumer := Consumer{}
ctx := context.Background()
for {
    err := ConsumerGroup.Consume(ctx, []string{topic}, Consumer)
    if err != nil {
        log.Panicf("Error from Consumer: %v", err)
    }
}

type Consumer struct {}
func (consumer Consumer) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (consumer Consumer) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (consumer Consumer) ConsumeClaim(sess sarama.ConsumerGroupSession, Claim sarama.ConsumerGroupClaim) error {
    for msg := range Claim.Messages() {
        fmt.Printf("Message topic:%q partition:%d offset:%d
", msg.Topic, msg.Partition, msg.Offset)
        sess.MarkMessage(msg, "")
    }
    return nil
}

Translator note: Although this article was published in May, the commit interval configuration shown is now deprecated; the current Sarama version uses Consumer.Offsets.AutoCommit instead of Consumer.Offsets.CommitInterval .

Minimizing Kafka Fetches

1. Increase Batch Size

Sending messages in larger batches improves throughput, though it may increase latency. In Go, adjust the producer’s batch size as follows:

config := sarama.NewConfig()
config.Producer.Flush.Bytes = 1024 * 1024

Similarly, increase the consumer’s fetch size:

config := sarama.NewConfig()
config.Consumer.Fetch.Default = 1024 * 1024

2. Use Long Polling

Long polling makes the consumer wait for data when the topic is empty, reducing unnecessary round‑trips.

config := sarama.NewConfig()
config.Consumer.MaxWaitTime = 500 * time.Millisecond

This configuration tells the consumer to wait up to 500 ms before returning an empty response.

Leverage Consumer Groups

Consumer groups allow multiple consumers to share the load of a topic. Sarama requires implementing the ConsumerGroupHandler interface:

type exampleConsumerGroupHandler struct {}
func (h *exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (h *exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h *exampleConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, Claim sarama.ConsumerGroupClaim) error {
    for message := range Claim.Messages() {
        fmt.Printf("Message: %s
", string(message.Value))
        session.MarkMessage(message, "")
    }
    return nil
}

Create and run the consumer group:

brokers := []string{"localhost:9092"}
topic := "example_topic"
groupID := "example_consumer_group"

consumerGroup, err := sarama.NewConsumerGroup(brokers, groupID, config)
if err != nil {
    log.Fatalf("Error creating consumer group: %v", err)
}
defer consumerGroup.Close()

handler := &exampleConsumerGroupHandler{}
for {
    err := consumerGroup.Consume(context.Background(), []string{topic}, handler)
    if err != nil {
        log.Printf("Error consuming messages: %v", err)
    }
}

The group can be scaled by adding more consumers, increasing overall processing capacity.

Adjust Consumer Buffer Size

Increasing the internal buffer lets the consumer hold more messages before processing, which can boost throughput at the cost of memory.

config := sarama.NewConfig()
config.Consumer.Return.Errors = true
config.Version = sarama.V2_1_0_0
config.Consumer.Offsets.Initial = sarama.OffsetOldest
config.ChannelBufferSize = 500

group, err := sarama.NewConsumerGroup([]string{broker}, groupID, config)
if err != nil { panic(err) }
ctx := context.Background()
for {
    topics := []string{topic}
    handler := exampleConsumerGroupHandler{}
    err := group.Consume(ctx, topics, &handler)
    if err != nil { panic(err) }
}

Handle Rebalance Events

When consumers join or leave a group, Kafka triggers a rebalance to redistribute partitions. Implementing Setup and Cleanup in the handler ensures graceful handling of these events, preventing duplicate processing or data loss.

Translator note: It is crucial to exit the ConsumeClaim loop promptly when the channel closes so that the Cleanup function runs correctly.

Monitor Consumers

Monitoring Kafka consumers is essential for health and performance. Key metrics include latency, processing time, and error rates. Useful tools and libraries:

Sarama’s built‑in metrics (exposed to Prometheus)

JMX Exporter for JVM‑based Kafka brokers

Kafka Exporter for detailed consumer‑group lag

Jaeger or OpenTelemetry for distributed tracing

Application logs for error detection kafka-consumer-groups CLI to inspect group state

Set up alerts on critical metrics to respond quickly to issues, ensuring robust, reliable, and efficient Kafka‑driven applications.