Go Client for Apache Kafka

confluent-kafka-go is the Go client for Apache Kafka® that Confluent develops and maintains. It includes a producer and a consumer, and it wraps librdkafka, the C/C++ client for Kafka, through cgo.

To build a Go client application step by step, see Getting Started with Apache Kafka and Go.

Install the Go client

Install the Go client with go get. On Linux (glibc and musl) and macOS, librdkafka is bundled with the Go client, so you don’t need to install it separately.

The Go client is available on GitHub and as the confluent-kafka-go package to pin to specific versions. The Changelog in the GitHub repository lists release updates.

On other platforms, first install the C client, including its development package, and a C build toolchain that includes pkg-config. The following table lists the additional packages for each platform:

Platform	Install command
Red Hat-based Linux distributions	sudo yum groupinstall "Development Tools"
Debian-based Linux distributions	sudo apt-get install build-essential pkg-config git
macOS with Homebrew	brew install pkg-config git

Install librdkafka before you run these commands on Red Hat-based and Debian-based distributions.

Next, use go get to install the library:

go get gopkg.in/confluentinc/confluent-kafka-go.v1/kafka

Your Go code can now import and use the client. To confirm that the installation succeeded, build and run the go-kafkacat command-line utility:

go get gopkg.in/confluentinc/confluent-kafka-go.v1/examples/go-kafkacat
$GOPATH/bin/go-kafkacat --help

The source code is also available in the ZIP and TAR archives under the directory src/.

Go client code examples

For a step-by-step tutorial that includes producer and consumer code samples, see Getting Started with Apache Kafka and Go. For more samples, see the Go client examples and the clients documentation.

Kafka producer

Create a producer

The Go client passes configuration to the producer in a ConfigMap object. A ConfigMap is a map of librdkafka configuration properties:

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

p, err := kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers": "host1:9092,host2:9092",
    "client.id": "myProducer",
    "acks": "all"})

if err != nil {
    fmt.Printf("Failed to create producer: %s\n", err)
    os.Exit(1)
}

Produce messages asynchronously

To produce a message asynchronously, call the Produce() method. Pass a Message object and, optionally, a chan Event that receives the result of the send. The Message object contains an opaque interface{} field that carries arbitrary data to the event handler:

err = p.Produce(&kafka.Message{
  TopicPartition: kafka.TopicPartition{Topic: topic, Partition: kafka.PartitionAny},
  Value: []byte(value)},
  nil, // delivery channel
)

Use a goroutine to handle delivery reports concurrently with other event types, such as errors and statistics. If you set a delivery channel, delivery reports arrive on that channel, and the client may start a goroutine to wait on them.

Always process the Events() channel, even when you set a delivery channel, because it contains client-level errors and control messages:

go func() {
    for e := range p.Events() {
        switch ev := e.(type) {
        case *kafka.Message:
            if ev.TopicPartition.Error != nil {
                fmt.Printf("Failed to deliver message: %v\n", ev.TopicPartition)
            } else {
                fmt.Printf("Successfully produced record to topic %s partition [%d] @ offset %v\n",
                    *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
            }
        }
    }
}()

Produce messages synchronously

To produce a message synchronously, receive from the delivery channel that you pass to the Produce() method. Synchronous writes reduce throughput, so use them only when your application requires them:

delivery_chan := make(chan kafka.Event, 10000)
err = p.Produce(&kafka.Message{
    TopicPartition: kafka.TopicPartition{Topic: topic, Partition: kafka.PartitionAny},
    Value: []byte(value)},
    delivery_chan
)

 e := <-delivery_chan
 m := e.(*kafka.Message)

 if m.TopicPartition.Error != nil {
     fmt.Printf("Delivery failed: %v\n", m.TopicPartition.Error)
 } else {
     fmt.Printf("Delivered message to topic %s [%d] at offset %v\n",
             *m.TopicPartition.Topic, m.TopicPartition.Partition, m.TopicPartition.Offset)
 }
 close(delivery_chan)

Alternatively, call the Flush() method to wait for all messages to be acknowledged:

p.Flush()

Important

Flush() serves only the producer’s Events() channel, not application-specified delivery channels. If you call Flush() and no goroutine processes the delivery channel, its buffer can fill up and cause the flush to time out.

Kafka consumer

Create a consumer

The Go client passes configuration to the consumer in a ConfigMap object. A ConfigMap is a map of librdkafka configuration properties:

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

consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
     "bootstrap.servers":    "host1:9092,host2:9092",
     "group.id":             "foo",
     "auto.offset.reset":    "smallest"})

Consume messages with Poll

A typical Kafka consumer application runs a consume loop that repeatedly calls the Consumer.Poll method. Poll retrieves records one by one from records that the consumer pre-fetches efficiently in background threads. Call Consumer.SubscribeTopics first to join a consumer group and choose the topics to consume from:

err = consumer.SubscribeTopics(topics, nil)

for run == true {
    ev := consumer.Poll(100)
    switch e := ev.(type) {
    case *kafka.Message:
        // application-specific processing
    case kafka.Error:
        fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
        run = false
    default:
        fmt.Printf("Ignored %v\n", e)
    }
}

consumer.Close()

Consumer.Poll returns as soon as a message or event is available, or after the timeout in milliseconds, whichever comes first. Choose the timeout as follows:

• A short value, such as 100, lets your application be interrupted.

• -1 waits indefinitely. Avoid it, because librdkafka cannot respond to SIGINT (the interrupt signal), so the application can’t be interrupted.

• 0 causes busy looping. Avoid it.

Call Consumer.Close when you finish using the consumer. Consumer.Close does the following:

• Closes active sockets and cleans up internal state.

• Triggers a group rebalance immediately, which reassigns the consumer’s partitions to another member of the group.

If you don’t close the consumer, the broker triggers the rebalance only after the session timeout expires.

Commit offsets synchronously

Call Commit() to commit offsets synchronously. Other commit methods accept more specific input:

• Commit() commits the current offsets.

• CommitOffsets() commits a list of offsets that you provide.

• CommitMessage() commits the offset relative to a consumed Message.

When you commit offsets manually, disable the enable.auto.commit configuration:

msg_count := 0
for run == true {
    ev := consumer.Poll(100)
    switch e := ev.(type) {
    case *kafka.Message:
        msg_count += 1
        if msg_count % MIN_COMMIT_COUNT == 0 {
            consumer.Commit()
        }
        fmt.Printf("%% Message on %s:\n%s\n",
            e.TopicPartition, string(e.Value))

    case kafka.PartitionEOF:
        fmt.Printf("%% Reached %v\n", e)
    case kafka.Error:
        fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
        run = false
    default:
        fmt.Printf("Ignored %v\n", e)
    }
}

This example triggers a synchronous commit every MIN_COMMIT_COUNT messages. You can also trigger the commit when a timeout expires, so that the committed position updates regularly.

At-least-once and at-most-once delivery

Committing after processing gives at-least-once delivery. If a failure occurs between processing a message and committing its offset, the consumer processes the message again. The synchronous commit example in the previous section works this way.

Committing before processing gives at-most-once delivery. If a failure occurs between the commit and processing, the message is lost, so handle commit failures carefully. The following example commits first, so it gives at-most-once delivery:

for run == true {
    ev := consumer.Poll(100)
    switch e := ev.(type) {
    case *kafka.Message:
        err = consumer.CommitMessage(e)
        if err == nil {
            msg_process(e)
        }

    case kafka.PartitionEOF:
        fmt.Printf("%% Reached %v\n", e)
    case kafka.Error:
        fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
        run = false
    default:
        fmt.Printf("Ignored %v\n", e)
    }
}

This example calls Consumer.CommitMessage() before processing the message for simplicity. Committing on every message adds significant overhead in practice. A better approach is to collect a batch of messages, run the synchronous commit, and then process the messages only if the commit succeeds.

Commit offsets asynchronously

To commit asynchronously, call Commit() in a goroutine:

msg_count := 0
for run == true {
    ev := consumer.Poll(100)
    switch e := ev.(type) {
    case *kafka.Message:
        msg_count += 1
        if msg_count % MIN_COMMIT_COUNT == 0 {
            go func() {
                offsets, err := consumer.Commit()
            }()
        }
        fmt.Printf("%% Message on %s:\n%s\n",
            e.TopicPartition, string(e.Value))

    case kafka.PartitionEOF:
        fmt.Printf("%% Reached %v\n", e)
    case kafka.Error:
        fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
        run = false
    default:
        fmt.Printf("Ignored %v\n", e)
    }
}

Handle rebalance events

To receive rebalance events, set go.application.rebalance.enable to true when you create the consumer. The Poll() method then returns rebalance events. Handle them as follows:

• On AssignedPartitions, call Assign().

• On RevokedPartitions, call Unassign().

consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
     "bootstrap.servers":    "host1:9092,host2:9092",
     "group.id":             "foo",
     "go.application.rebalance.enable": true})

msg_count := 0
for run == true {
    ev := consumer.Poll(100)
    switch e := ev.(type) {
    case kafka.AssignedPartitions:
        fmt.Fprintf(os.Stderr, "%% %v\n", e)
        c.Assign(e.Partitions)
    case kafka.RevokedPartitions:
        fmt.Fprintf(os.Stderr, "%% %v\n", e)
        c.Unassign()
    case *kafka.Message:
        msg_count += 1
        if msg_count % MIN_COMMIT_COUNT == 0 {
            consumer.Commit()
        }

        fmt.Printf("%% Message on %s:\n%s\n",
            e.TopicPartition, string(e.Value))

    case kafka.PartitionEOF:
        fmt.Printf("%% Reached %v\n", e)
    case kafka.Error:
        fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
        run = false
    default:
        fmt.Printf("Ignored %v\n", e)
    }
}

Go client API reference

See the Go client API documentation.

Related content

• Client quick start for Confluent Cloud (Confluent Documentation)

• Getting Started with Apache Kafka and Go (Confluent Developer)

• Apache Kafka 101 (Confluent Developer)

• Schema Registry Client for Go (Confluent GitHub)