franz-go - A complete 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 (Kafka >= 0.8.0 through v3.4+)
• Full Exactly-Once-Semantics (EOS)
• Idempotent & transactional producers
• Simple (legacy) consumer
• Group consumers with eager (roundrobin, range, sticky) and cooperative (cooperative-sticky) balancers
• All compression types supported: gzip, snappy, lz4, zstd
• SSL/TLS provided through custom dialer options
• All SASL mechanisms supported (GSSAPI/Kerberos, PLAIN, SCRAM, and OAUTHBEARER)
• Low-level admin functionality supported through a simple Request function
• High-level admin package with many helper types to make cluster administration easy.
• Utilizes modern & idiomatic Go (support for contexts, variadic configuration options, ...)
• Highly performant by avoiding channels and goroutines where not necessary
• Written in pure Go (no wrapper lib for a C library or other bindings)
• Ability to add detailed log messages or metrics using hooks
• Plug-in metrics support for prometheus, zap, etc.
• An admin client with many helper functions for easy admin tasks
• A schema registry client and convenience Serde type for encoding and decoding

Works with any Kafka compatible brokers:

• Redpanda: the fastest and most efficient Kafka compatible event streaming platform
• Kafka: the original Java project
• Microsoft Event Hubs
  • Event Hubs does not support producing with compression; be sure to use kgo.ProducerBatchCompression(kgo.NoCompression).
• Amazon MSK

Install

This repo contains multiple tags to allow separate features to be developed and
released independently. The main client is in franz-go. Plugins are released
from plugin/{plugin}. The raw-protocol package is released from pkg/kmsg,
and the admin package is released from pkg/kadm.

The main client is located in the package github.com/twmb/franz-go/pkg/kgo,
while the root of the project is at github.com/twmb/franz-go. There are
a few extra packages within the project, as well as a few sub-modules. To
use the main kgo package,

go get github.com/twmb/franz-go

To use a plugin,

go get github.com/twmb/franz-go/plugin/kzap

To use kadm,

go get github.com/twmb/franz-go/pkg/kadm

As an example, your require section in go.mod may look like this:

require (
	github.com/twmb/franz-go v1.12.0
	github.com/twmb/franz-go/pkg/kmsg v1.4.0
)

Getting started

Here's a basic overview of producing and consuming:

seeds := []string{"localhost:9092"}
// One client can both produce and consume!
// Consuming can either be direct (no consumer group), or through a group. Below, we use a group.
cl, err := kgo.NewClient(
	kgo.SeedBrokers(seeds...),
	kgo.ConsumerGroup("my-group-identifier"),
	kgo.ConsumeTopics("foo"),
)
if err != nil {
	panic(err)
}
defer cl.Close()

ctx := context.Background()

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

})
wg.Wait()

// Alternatively, ProduceSync exists to synchronously produce a batch of records.
if err := cl.ProduceSync(ctx, record).FirstErr(); err != nil {
	fmt.Printf("record had a produce error while synchronously producing: %v\n", err)
}

// 2.) Consuming messages from a topic
for {
	fetches := cl.PollFetches(ctx)
	if errs := fetches.Errors(); len(errs) > 0 {
		// All errors are retried internally when fetching, but non-retriable errors are
		// returned from polls so that users can notice and take action.
		panic(fmt.Sprint(errs))
	}

	// We can iterate through a record iterator...
	iter := fetches.RecordIter()
	for !iter.Done() {
		record := iter.Next()
		fmt.Println(string(record.Value), "from an iterator!")
	}

	// or a callback function.
	fetches.EachPartition(func(p kgo.FetchTopicPartition) {
		for _, record := range p.Records {
			fmt.Println(string(record.Value), "from range inside a callback!")
		}

		// We can even use a second callback!
		p.EachRecord(func(record *kgo.Record) {
			fmt.Println(string(record.Value), "from a second callback!")
		})
	})
}

This only shows producing and consuming in the most basic sense, and does not
show the full list of options to customize how the client runs, nor does it
show transactional producing / consuming. Check out the examples
directory for more!

API reference documentation can be found on
.
Supplementary information can be found in the docs directory:

Who uses this?

In alphabetical order,

• Alpaca
• Banyan
• Eoitek
• Mux
• Redpanda Console
• Redpanda Data
• StoneCo
• ThinkingData
• Unistack (Cloud Management System)
• Unity Technologies

If you use this library and want on the list above, please either open a PR
or comment on #142!

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. If you want to pin to an exact version, you can use the MaxVersions
option.

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.

Metrics & logging

Note there exists plug-in packages that allow you to easily add prometheus
metrics, go-metrics, zap logging, etc. to your client! See the plugin
directory for more information! These plugins are provided under dedicated
modules, e.g. github.com/twmb/franz-go/plugin/kprom@v1.0.0.

The franz-go client takes a neutral approach to metrics by providing hooks
that you can use to plug in your own metrics.

All connections, disconnections, reads, writes, and throttles can be hooked
into, as well as per-batch produce & consume metrics. If there is an aspect of
the library that you wish you could have insight into, please open an issue and
we can discuss adding another hook.

Hooks allow you to log in the event of specific errors, or to trace latencies,
count bytes, etc., all with your favorite monitoring systems.

In addition to hooks, logging can be plugged in with a general Logger
interface. A basic logger is provided if you just want to write to a given
file in a simple format. All logs have a message and then key/value pairs of
supplementary information. It is recommended to always use a logger and to use
LogLevelInfo.

See this example for an expansive example of
integrating with prometheus! Alternatively, see this example
for how to use the plug-in prometheus package!

Benchmarks

This client is quite fast; it is the fastest and most cpu and memory efficient
client in Go.

For 100 byte messages,

• This client is 4x faster at producing than confluent-kafka-go, and up to
  10x-20x faster (at the expense of more memory usage) at consuming.

• This client is 2.5x faster at producing than sarama, and 1.5x faster at
  consuming.

• This client is 2.4x faster at producing than segment's kafka-go, and so
  much faster at consuming that I'm not sure I wrote the consuming comparison
  correctly here.

To check benchmarks yourself, see the bench example. This
example lets you produce or consume to a cluster and see the byte / record
rate. The compare subdirectory shows comparison
code.

Supported KIPs

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