<a id="go-client"></a>

# Go Client for Apache Kafka

Confluent develops and maintains a Go client for Apache Kafka® that offers a producer
and a consumer. This library provides capabilities to produce to and consume
from Kafka topics using Go.

For a step-by-step guide on building a Go client application for Kafka, see [Getting Started with Apache Kafka and Go](https://developer.confluent.io/get-started/go/).

<a id="installation-go-client"></a>

## Installation

The Go client, called confluent-kafka-go, is distributed via [GitHub](https://github.com/confluentinc/confluent-kafka-go)
and as [confluent-kafka-go](https://gopkg.in/confluentinc/confluent-kafka-go.v1/kafka) to pin to specific versions.
The [Changelog](https://github.com/confluentinc/confluent-kafka-go/blob/master/CHANGELOG.md) showing release updates
is available in that same repo. The Go client uses librdkafka, the C client,
internally and exposes it as Go library using [cgo](https://golang.org/cmd/cgo/).
Starting with confluent-kafka-go v1.4.0, the librdkafka client is now
included in the Go client and no separate installation of librdkafka is
required for the supported platforms (Linux (glibc and musl based), and Mac OSX).

For other platforms the following instructions still apply:
To install the Go client, first install
[the C client](https://docs.confluent.io/kafka-clients/librdkafka/current/overview.html) including its development package as well as a C build toolchain including
`pkg-config`. On Red Hat-based Linux distributions install the following packages in addition to librdkafka:

```bash
sudo yum groupinstall "Development Tools"
```

On Debian-based distributions, install the following in addition to librdkafka:

```bash
sudo apt-get install build-essential pkg-config git
```

On macOS using [Homebrew](http://brew.sh/), install the following:

```bash
brew install pkg-config git
```

Next, use `go get` to install the library:

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

Your Go code can now import and use the client. You can also build and run a small command line utility, `go-kafkacat`,
to ensure the installation was successful:

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

See the [clients documentation](/platform/current/tutorials/examples/clients/docs/clients-all-examples.html) for code examples showing how to use the library.

The source code is also available in the [ZIP and TAR archives](/platform/current/installation/installing_cp/zip-tar.html) under the directory `src/`.

## Example code

For a step-by-step tutorial using the Go client including code samples for the producer and consumer see [this guide](https://developer.confluent.io/get-started/go/).

There are also additional examples [here](/platform/current/tutorials/examples/clients/docs/go.html#client-examples-go).

<a id="producer-go-client"></a>

## Kafka Producer

### Initialization

The Go client uses a `ConfigMap` object to pass configuration to the producer:

```bash
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)
}
```

### Asynchronous writes

In Go, you initiate a send by calling the `Produce()` method, passing a `Message` object and an
optional `chan Event` that can be used to listen for the result of the send. The `Message`
object contains an opaque `interface{}` field that can be used to pass arbitrary data with the
message to the subsequent event handler:

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

To produce asynchronously, you can use a goroutine to handle message delivery reports and optionally
other event types such as errors, statistics, and more, concurrently. If a delivery channel is set,
the message delivery reports will be delivered on that channel, and a goroutine may be spawned
specifically to wait on them. However, even in this case, the Events() channel must be processed,
because it contains client-level errors and control messages:

```bash
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)
            }
        }
    }
}()
```

### Synchronous writes

Making writes synchronous is typically a bad idea since it decreases throughput.
Nevertheless, you can make writes synchronous by receiving from the delivery channel
passed to the `Produce()` method call as shown below:

```bash
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, you can wait for all messages to be acknowledged and use the `Flush()` method:

```bash
p.Flush()
```

#### IMPORTANT
`Flush()` will only serve the producer’s `Events()` channel, not
application-specified delivery channels. If `Flush()` is called and no goroutine
is processing the delivery channel, its buffer may fill up and cause the
flush to timeout.

<a id="consumer-go-client"></a>

## Kafka Consumer

### Initialization

The Go client uses a `ConfigMap` object to pass configuration to the consumer:

```go
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"})
```

### Basic usage

A typical Kafka consumer application is centered around a consume loop, which
repeatedly calls the `Consumer.Poll` method to retrieve records one-by-one that
have been efficiently pre-fetched by the consumer in background threads:

```go
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()
```

The `Consumer.Poll` method returns as soon as a message or event is available, or after the specified
timeout (in milliseconds), whichever is sooner. You can specify a timeout of `-1` to indicate
an infinite timeout, however since `librdkafka` is not yet able to respond to SIGINT events,
it is advisable to specify a short timeout to allow your application to be interrupted. To avoid
busy looping, you should not specify a timeout of `0`.

The `Consumer.SubscribeTopics` method is used to join a consumer group and specify which
topics to consume from.

You should always call `Consumer.Close` after you are finished
using the consumer. Doing so will ensure that active sockets are
closed and internal state is cleaned up. It will also trigger a group
rebalance immediately which ensures that any partitions owned by the
consumer are re-assigned to another member in the group. If not closed
properly, the broker will trigger the rebalance only after the session
timeout has expired. Latch is added to this example to ensure
that the consumer has time to finish closing before finishing
shut down.

#### Synchronous commits

The Go client provides a synchronous `Commit()`
method call. Other variants of commit methods also accept a list of offsets to
commit or a `Message` in order to commit offsets relative to a consumed message.
When using manual offset commit, be sure to disable the `enable.auto.commit` configuration.

```go
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)
    }
}
```

In this example, a synchronous commit is triggered every `MIN_COMMIT_COUNT`
messages. You could also trigger the commit on expiration of a
timeout to ensure there the committed position is updated regularly.

#### Delivery guarantees

In the previous example, you get “at least once”
delivery since the commit follows the message processing. By changing the order,
however, you can get “at most once” delivery, but you must be a
little careful with the commit failure.

```go
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)
    }
}
```

For simplicity in this example, `Consumer.CommitMessage()` is used
prior to processing the message. Committing on every message would
produce a lot of overhead in practice. A better approach would be to
collect a batch of messages, execute the synchronous commit, and then
process the messages only if the commit succeeded.

#### Asynchronous commits

To commit asynchronously, simply execute the commit in a goroutine to commit asynchronously:

```go
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)
    }
}
```

Rebalance events are exposed as events returned by the `Poll()`
method. To see these events you must create the consumer with the
`go.application.rebalance.enable` configuration and handle
`AssignedPartitions` and `RevokedPartitions` events by explicitly calling
`Assign()` and `Unassign()` for `AssignedPartitions` and
`RevokedPartitions` respectively:

```go
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)
    }
}
```

## API documentation

Click [here](/platform/current/clients/confluent-kafka-go/index.html) to view the Go Client API documentation.

## Related content

* [Client quick start for Confluent Cloud](https://docs.confluent.io/cloud/current/client-apps/config-client.html) (Confluent Documentation)
* [Getting Started with Apache Kafka and Go](https://developer.confluent.io/get-started/go/) (Confluent Developer)
* [Apache Kafka 101](https://developer.confluent.io/learn-kafka/apache-kafka/) (Confluent Developer)
* [Schema Registry Client for Go](https://github.com/confluentinc/confluent-kafka-go) (Confluent GitHub)