franz-go - Apache Kafka client written in Go

Franz-go is an all-encompassing Apache Kafka client fully written Go. This library aims to provide every Kafka feature from Apache Kafka v0.8.0 onward. It has support for transactions, regex topic consuming, the latest partitioning strategies, data loss detection, closest replica fetching, and more. If a client KIP exists, this library aims to support it.

This library attempts to provide an intuitive API while interacting with Kafka the way Kafka expects (timeouts, etc.).

Features

Feature complete client (up to Kafka v2.7.0+)
Supported compression types: snappy, gzip, lz4 and zstd
SSL/TLS Support
Exactly once semantics / idempotent producing
Transactions support
All SASL mechanisms are supported (OAuthBearer, GSSAPI/Kerberos, SCRAM-SHA-256/512 and plain)
Supported Kafka versions >=0.8
Provides low level functionality (such as sending API requests) as well as high level functionality (e.g. consuming in groups)
Utilizes modern & idiomatic Go (support for contexts, variadic configuration options, ...)
Highly performant, see Performance (benchmarks will be added)
Written in pure Go (no wrapper lib for a C library or other bindings)
Ability to add detailed log messages or metrics using hooks

Getting started

Basic usage for producing and consuming Kafka messages looks like this:

seeds := []string{"localhost:9092"}
client, err := kgo.NewClient(kgo.SeedBrokers(seeds...))
if err != nil {
    panic(err)
}
defer client.Close()

ctx := context.Background()

// 1.) Producing a message
// All record production goes through Produce, and the callback can be used
// to allow for syncronous or asyncronous production.
var wg sync.WaitGroup
wg.Add(1)
record := &kgo.Record{Topic: "foo", Value: []byte("bar")}
err := client.Produce(ctx, record, func(_ *Record, err error) {
        defer wg.Done()
        if err != nil {
                fmt.Printf("record had a produce error: %v\n", err)
        }

}
if err != nil {
        panic("we are unable to produce if the context is canceled, we have hit max buffered," +
                "or if we are transactional and not in a transaction")
}
wg.Wait()

// 2.) Consuming messages from a topic
// Consuming can either be direct (no consumer group), or through a group. Below, we use a group.
// client.AssignGroup("my-group-identifier", kgo.GroupTopics("foo"))
for {
        fetches := client.PollFetches(ctx)
        iter := fetches.RecordIter()
        for !iter.Done() {
            record := iter.Next()
            fmt.Println(string(record.Value))
        }
}

Take a look at more examples
Architecture documentation describing all Franz-go package purposes
Reference docs https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo
Consuming in Consumer Groups
Producing
Sending admin requests

Version Pinning

By default, the client issues an ApiVersions request on connect to brokers and defaults to using the maximum supported version for requests that each broker supports.

Kafka 0.10.0 introduced the ApiVersions request; if you are working with brokers older than that, you must use the kversions package. Use the MaxVersions option for the client if you do so.

As well, it is recommended to set the MaxVersions to the version of your broker cluster. Until KIP-584 is implemented, it is possible that if you do not pin a max version, this client will speak with some features to one broker while not to another when you are in the middle of a broker update roll.

Metrics

Using hooks you can attach to any events happening within franz-go. This allows you to use your favorite metric library and collect the metrics you are interested in.

Supported KIPs

Theoretically, this library supports every (non-Java-specific) client facing KIP. Most are tested, some need testing. Any KIP that simply adds or modifies a protocol is supported by code generation.