Working with Apache Kafka in Go (Golang): Libraries and Examples

Working with Apache Kafka in Go (Golang): Libraries, Examples, and Best Practices

Apache Kafka is a distributed streaming platform that has become the de facto standard for building high-load messaging systems. In the Go (Golang) ecosystem, there are several mature libraries for interacting with Kafka. In this article, we will cover the two most popular ones: confluent-kafka-go (the official client from Confluent) and sarama (a pure Go client). You will learn how to set up a producer and consumer, handle errors, and choose the right tool for your project.

Why Kafka and Go are a Great Pair?

Go was designed for building fast, concurrent, and reliable server applications. Kafka, in turn, requires high throughput and minimal latency. Go goroutines and channels are ideal for asynchronous message stream processing. Additionally, Go's static typing helps avoid many errors at compile time, which is critical for production systems.

1. The confluent-kafka-go Library

This is a wrapper around the C client librdkafka, providing maximum performance and full support for all Kafka features (transactions, idempotence, security protocols). It is recommended for high-load systems.

Installation

Requires librdkafka to be installed on the system:

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# Then install the Go packagego get -u github.com/confluentinc/confluent-kafka-go/kafka

Producer Example

Sends a message to the orders topic:

package main

import (    "fmt"    "github.com/confluentinc/confluent-kafka-go/kafka")

func main() {    p, err := kafka.NewProducer(&kafka.ConfigMap{        "bootstrap.servers": "localhost:9092",        "acks":              "all",    })    if err != nil {        panic(err)    }    defer p.Close()

    // Channel for delivery notifications    go func() {        for e := range p.Events() {            switch ev := e.(type) {            case *kafka.Message:                if ev.TopicPartition.Error != nil {                    fmt.Printf("Delivery error: %v", ev.TopicPartition.Error)                } else {                    fmt.Printf("Message delivered to %s [%d] offset %d",                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)                }            }        }    }()

    topic := "orders"    message := []byte("New order #12345")    err = p.Produce(&kafka.Message{        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},        Value:          message,        Key:            []byte("order_key"),    }, nil)

    if err != nil {        fmt.Printf("Send error: %v", err)    }

    // Wait for all messages to be sent    p.Flush(15 * 1000)}

Consumer Example

Reads messages from the orders topic in the order-processors group:

package main

import (    "fmt"    "github.com/confluentinc/confluent-kafka-go/kafka")

func main() {    c, err := kafka.NewConsumer(&kafka.ConfigMap{        "bootstrap.servers": "localhost:9092",        "group.id":          "order-processors",        "auto.offset.reset": "earliest",    })    if err != nil {        panic(err)    }    defer c.Close()

    err = c.SubscribeTopics([]string{"orders"}, nil)    if err != nil {        panic(err)    }

    for {        msg, err := c.ReadMessage(-1) // infinite wait        if err != nil {            fmt.Printf("Receive error: %v", err)            continue        }        fmt.Printf("Received message: %s (key: %s) from %s [%d] of