franz-go

README

franz-go

A complete Apache Kafka client written in Go

Franz-go is an all-encompassing Apache Kafka client fully written in 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 v4.4+)
• Full Exactly-Once-Semantics (EOS)
• Idempotent & transactional producers
• Simple (legacy) consumer
• Group consumers with eager (roundrobin, range, sticky) and cooperative (cooperative-sticky) balancers, with optional rack-aware partition assignment (KIP-881) or server-side balancing (KIP-848)
• Share group (queue) consumers (KIP-932)
• 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
• A 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.
• A schema registry client and convenience Serde type for encoding and decoding

Works with any Kafka compatible brokers:

• Kafka
• Redpanda
• Confluent Platform
• 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.21.0
	github.com/twmb/franz-go/pkg/kmsg v1.13.1
)

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:

docs
├── admin requests — an overview of how to issue admin requests
├── metrics and logging — a small writeup on how to enable metrics & logging in franz-go, as well as a few thoughts on latency tracking
├── package layout — describes the packages in franz-go
├── producing and consuming — descriptions of producing & consuming & the guarantees
├── transactions — a description of transactions and the safety even in a pre-KIP-447 world
└── testing — describes how to write tests for your code that uses franz-go

Who uses this?

In alphabetical order,

• Abnormal AI
• Alpaca
• Artie
• Banyan
• Benthos
• Conduit
• DeltaStream
• EasyPost
• Eoitek
• Hilton
• Mux
• Redpanda Console
• Redpanda Data
• Sharechat
• StoneCo
• ThinkingData
• Unistack (Cloud Management System)
• Unity Technologies
• Zomato

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.

Topic recreation

The client always produces to or consumes from the first instance of a topic. If you delete and recreate a topic, and if the client can detect it (i.e. the broker is modern enough and is using topic IDs), the client will refuse to produce to or consume from the new instance of the topic.

You can handle this in your application by using PurgeTopicsFromClient, which purges all information about a topic from within the client, and then (for consumers) calling AddConsumeTopics. Producers will auto recover after purging once you produce again.

Depending on your broker version, you may experience different side effects from trying to keep a client alive across topic recreation:

• Producing pre-4.1 while recreating a topic may result in batch 2 being before batch 1 on the new topic, and may result in some duplicates.
• Consuming pre-3.1 could result in the new topic's messages being skipped, since the consumer will stay at the offset it was at in the old topic.
• Committing pre-4.2 could result in commits for the old topic landing for the new topic, and consumers picking up these stale commits and skipping messages
• Transactional commits pre-4.4 have the same problem.

It is not recommended to try to keep a client alive across a topic recreation, but if you want to do so: always upgrade to the latest Kafka. Doing so will make this client hard-lock more reliably and avoid side effects; you can then purge and add as needed.

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.3.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!

Performance

This client is fast, and is consistently among the fastest and most cpu and memory efficient Kafka clients in Go. To measure yourself, see the bench example; the compare subdirectory has code for comparing against other clients.

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.

KIP	Kafka release	Status
KIP-1 — Disallow acks > 1	0.8.3	Supported & Enforced
KIP-4 — Request protocol changes	0.9.0 through 0.10.1	Supported
KIP-8 — Flush method on Producer	0.8.3	Supported
KIP-12 — SASL & SSL	0.9.0	Supported
KIP-13 — Throttling (on broker)	0.9.0	Supported
KIP-15 — Close with a timeout	0.9.0	Supported (via context)
KIP-19 — Request timeouts	0.9.0	Supported
KIP-22 — Custom partitioners	0.9.0	Supported
KIP-31 — Relative offsets in message sets	0.10.0	Supported
KIP-32 — Timestamps in message set v1	0.10.0	Supported
KIP-35 — ApiVersion	0.10.0	Supported
KIP-40 — ListGroups and DescribeGroups	0.9.0	Supported
KIP-41 — max.poll.records	0.10.0	Supported (via PollRecords)
KIP-42 — Producer & consumer interceptors	0.10.0	Supported via hooks
KIP-43 — SASL PLAIN & handshake	0.10.0	Supported
KIP-48 — Delegation tokens	1.1	Supported
KIP-54 — Sticky partitioning	0.11.0	Supported
KIP-57 — Fix lz4	0.10.0	Supported
KIP-62 — background heartbeats & improvements	0.10.1	Supported
KIP-70 — On{Assigned,Revoked}	0.10.1	Supported
KIP-74 — Fetch response size limits	0.10.1	Supported
KIP-78 — ClusterID in Metadata	0.10.1	Supported
KIP-79 — List offsets for times	0.10.1	Supported
KIP-81 — Bound fetch memory usage	WIP	Supported (through options)
KIP-82 — Record headers	0.11.0	Supported
KIP-84 — SASL SCRAM	0.10.2	Supported
KIP-86 — SASL Callbacks	0.10.2	Supported (through callback fns)
KIP-88 — OffsetFetch for admins	0.10.2	Supported
KIP-91 — Intuitive producer timeouts	2.1	Supported (as a matter of opinion)
KIP-97 — Backwards compat for old brokers	0.10.2	Supported
KIP-98 — EOS	0.11.0	Supported
KIP-101 — OffsetForLeaderEpoch v0	0.11.0	Supported
KIP-102 — Consumer close timeouts	0.10.2	Supported (via context)
KIP-107 — DeleteRecords	0.11.0	Supported
KIP-108 — CreateTopic validate only field	0.10.2	Supported
KIP-110 — zstd	2.1	Supported
KIP-112 — Broker request protocol changes	1.0	Supported
KIP-113 — LogDir requests	1.0	Supported
KIP-117 — Admin client	0.11.0	Supported
KIP-124 — Request rate quotas	0.11.0	Supported
KIP-126 — Ensure proper batch size after compression	0.11.0	Supported (avoided entirely)
KIP-133 — Describe & Alter configs	0.11.0	Supported
KIP-140 — ACLs	0.11.0	Supported
KIP-144 — Broker reconnect backoff	0.11.0	Supported
KIP-152 — More SASL; SASLAuthenticate	1.0	Supported
KIP-183 — Elect preferred leaders	2.2	Supported
KIP-185 — Idempotency is default	1.0	Supported
KIP-192 — Cleaner idempotence semantics	1.0	Supported
KIP-195 — CreatePartitions	1.0	Supported
KIP-204 — DeleteRecords via admin API	1.1	Supported
KIP-207 — New error in ListOffsets	2.2	Supported
KIP-219 — Client-side throttling	2.0	Supported
KIP-222 — Group operations via admin API	2.0	Supported
KIP-226 — Describe configs v1	1.1	Supported
KIP-227 — Incremental fetch	1.1	Supported
KIP-229 — DeleteGroups	1.1	Supported
KIP-249 — Delegation tokens in admin API	2.0	Supported
KIP-255 — SASL OAUTHBEARER	2.0	Supported
KIP-266 — Fix indefinite consumer timeouts	2.0	Supported (via context)
KIP-279 — OffsetForLeaderEpoch bump	2.0	Supported
KIP-289 — Default group.id to null	2.2	Supported
KIP-294 — TLS verification	2.0	Supported (via dialer)
KIP-302 — Use multiple addrs for resolved hostnames	2.1	Supported (via dialer)
KIP-320 — Fetcher: detect log truncation	2.1	Supported
KIP-322 — DeleteTopics disabled error code	2.1	Supported
KIP-339 — IncrementalAlterConfigs	2.3	Supported
KIP-341 — Sticky group bugfix	?	Supported
KIP-342 — OAUTHBEARER extensions	2.1	Supported
KIP-345 — Static group membership	2.4	Supported
KIP-357 — List ACLs per principal via admin API	2.1	Supported
KIP-360 — Safe epoch bumping for UNKNOWN_PRODUCER_ID	2.5	Supported
KIP-361 — Allow disable auto topic creation	2.3	Supported
KIP-368 — Periodically reauthenticate SASL	2.2	Supported
KIP-369 — An always round robin produce partitioner	2.4	Supported
KIP-373 — Users can create delegation tokens for others	3.3	Supported
KIP-380 — Inter-broker protocol changes	2.2	Supported
KIP-389 — Group max size error	2.2	Supported
KIP-390 — Configurable compression level	2.1	Supported (via CompressionCodec.WithLevel)
KIP-392 — Closest replica fetching w/ rack	2.2	Supported
KIP-394 — Require member.id for initial join request	2.2	Supported
KIP-396 — Commit offsets manually	2.4	Supported
KIP-405 — Kafka Tiered Storage	3.5	Supported (protos)
KIP-412 — Dynamic log levels w/ IncrementalAlterConfigs	2.4	Supported
KIP-429 — Incremental rebalance (see KAFKA-8179)	2.4	Supported
KIP-430 — Authorized ops in DescribeGroups	2.3	Supported
KIP-447 — Producer scalability for EOS	2.5	Supported
KIP-455 — Replica reassignment API	2.4	Supported
KIP-460 — Leader election API	2.4	Supported
KIP-464 — CreateTopic defaults	2.4	Supported
KIP-467 — Per-record error codes when producing	2.4	Supported (and ignored)
KIP-480 — Sticky partition producing	2.4	Supported
KIP-482 — Tagged fields (KAFKA-8885)	2.4	Supported
KIP-487 — Client-side auto topic creation control	?	Supported (via AllowAutoTopicCreation)
KIP-496 — OffsetDelete admin command	2.4	Supported
KIP-497 — New AlterISR API	2.7	Supported
KIP-498 — Max bound on reads	?	Supported
KIP-511 — Client name/version in ApiVersions request	2.4	Supported
KIP-514 — Bounded Flush	2.4	Supported (via context)
KIP-516 — Topic IDs	2.8	Supported
KIP-518 — List groups by state	2.6	Supported
KIP-519 — Configurable SSL "engine"	2.6	Supported (via dialer)
KIP-525 — CreateTopics v5 returns configs	2.4	Supported
KIP-526 — Reduce metadata lookups	2.5	Supported
KIP-533 — Default API timeout (total time, not per request)	2.5	Supported (via RetryTimeout)
KIP-546 — Client Quota APIs	2.5	Supported
KIP-554 — Broker side SCRAM APIs	2.7	Supported
KIP-559 — Protocol info in sync/join	2.5	Supported
KIP-568 — Explicit rebalance triggering on the consumer	2.6	Supported
KIP-569 — Docs & type in DescribeConfigs	2.6	Supported
KIP-570 — Leader epoch in StopReplica	2.6	Supported
KIP-580 — Exponential backoff	2.6	Supported
KIP-584 — Versioning scheme for features	2.6	Supported
KIP-588 — Producer recovery from txn timeout	2.7	Supported
KIP-590 — Envelope (broker only)	2.7	Supported
KIP-595 — New APIs for raft protocol	2.7	Supported
KIP-599 — Throttling on create/delete topic/partition	2.7	Supported
KIP-601 — Configurable socket connection timeout	2.7	Supported (via dialer)
KIP-602 — Use all resolved addrs by default	2.6	Supported (via dialer)
KIP-651 — Support PEM	2.7	Supported (via dialer)
KIP-654 — Aborted txns with unflushed data is not fatal	2.7	Supported (default behavior)
KIP-664 — Describe producers / etc.	2.8 (mostly)	Supported
KIP-679 — Strongest producer guarantee by default	3.0	Supported (by default always)
KIP-699 — Batch FindCoordinators	3.0	Supported
KIP-700 — DescribeCluster	2.8	Supported
KIP-704 — AlterISR => AlterPartition	3.2	Supported
KIP-709 — Batch OffsetFetch	3.0	Supported
KIP-714 — Client Metrics	3.7	Supported
KIP-730 — AllocateProducerIDs	3.0	Supported
KIP-734 — Support MaxTimestamp in ListOffsets	3.0	Supported (LookbackOffset)
KIP-735 — Bump default session timeout	?	Supported
KIP-768 — Extend SASL/OAUTHBEARER support for OIDC	3.1	Supported
KIP-778 — KRaft Upgrades (protocol changes only)	3.2	Supported
KIP-784 — Add ErrorCode to DescribeLogDirs response	3.1	Supported
KIP-792 — Generation field in consumer group protocol	3.4	Supported
KIP-794 — Better sticky partitioning	3.3	Supported (UniformBytesPartitioner)
KIP-800 — Reason in Join/Leave group	3.1	Supported
KIP-814 — SkipAssignment for static group leaders	3.1	Supported
KIP-827 — DescribeLogDirs.{Total,Usable}Bytes	3.3	Supported
KIP-836 — DescribeQuorum voter lag info	3.3	Supported
KIP-841 — AlterPartition.TopicID	3.3	Supported
KIP-848 — Next gen consumer rebalance protocol	3.7	Supported (ServerSideBalancer group opt)
KIP-851 — RequireStable on OffsetFetch	3.3	Supported (via kadm.RequireStable)
KIP-853 — Add replica directory ID for replica fetchers	3.9	Supported
KIP-858 — JBOD in KRaft (protocol)	3.7	Supported
KIP-860 — Client side AlterPartitionAssignments RF change guard	4.1	Supported (kadm v1.17+)
KIP-866 — ZK to Raft RPC changes	3.4	Supported
KIP-881 — Rack-aware consumer partition assignment	3.5	Supported (range & sticky)
KIP-890 — Transactions server side defense	3.8, 4.0	Supported
KIP-893 — Nullable structs in the protocol	3.5	Supported
KIP-899 — Allow clients to rebootstrap	?	Supported (UpdateSeedBrokers)
KIP-909 — DNS resolution failure does not fail the client	4.4	Supported (existing)
KIP-903 — Stale broker epoch fencing	3.5	Supported (proto)
KIP-919 — Admin client to KRaft, Controller registration	3.7	Supported
KIP-932 — Share groups (queues)	4.1	Supported (via ShareGroup)
KIP-951 — Leader discovery optimizations	3.7	Supported
KIP-966 — Eligible leader replicas (protocol)	3.7	Supported
KIP-994 — List/Describe transactions enhancements	3.8 (partial)	Supported
KIP-1000 — ListClientMetricsResources	3.7	Supported
KIP-1005 — ListOffsets w. Timestamp -5	3.9	Supported
KIP-1022 — Formatting changes for features	4.0	Supported
KIP-1023 — ListOffsets earliest pending upload offset	4.2	Supported (existing)
KIP-1025 — URL-encode clientID/secret in OAuth auth header	3.9	Supported (user callback)
KIP-1043 — Administration of groups	4.0	Supported
KIP-1066 — Cordoned brokers and log directories	4.3	Supported (kadm)
KIP-1071 — Streams group protocol	4.2	Supported (proto)
KIP-1073 — DescribeCluster.IsFenced	4.0	Supported
KIP-1075 — TimeoutMillis on ListOffsets	4.0	Supported
KIP-1076 — User provided client metrics	4.0	Supported
KIP-1082 — ClientID in the next-gen rebalancer	4.0	Supported
KIP-1102 — RebootstrapRequired	4.0	Supported
KIP-1106 — Duration-based offset reset	4.0	Supported (LookbackOffset)
KIP-1123 — Rack-aware producer partitioning	4.4	Supported
KIP-1139 — Oauth JWT bearer grant	4.0	Supported
KIP-1142 — ListConfigResources	4.1	Supported (kadm v1.17+)
KIP-1152 — ListTransactions.TransactionalIDPattern	4.1	Supported (kadm v1.17+)
KIP-1160 — Per-broker supported features in ApiVersions	4.2	Supported (proto)
KIP-1166 — Improve HWM replication (protocol)	4.1	Supported
KIP-1186 — AddRaftVoter AckWhenCommitted	4.2	Supported (proto)
KIP-1206 — ShareFetch strict record limit	4.2	Supported (via ShareMaxRecordsStrict)
KIP-1222 — Share consumer renew acknowledgements	4.2	Supported (via AckRenew)
KIP-1226 — Share partition lag in DescribeShareGroupOffsets	4.2	Supported (kadm)
KIP-1227 — Rack ID in (Share)MemberDescription	4.2	Supported (kadm)
KIP-1228 — WriteTxnMarkers TransactionVersion	4.2	Supported
KIP-1242 — Detection and handling of misrouted connections	4.4	Supported
KIP-1258 — OAuth client assertion in client_credentials grant	4.3	Supported (user callback)
KIP-1319 — TxnOffsetCommit by topic ID	4.4	Supported
KIP-1331 — Streams group topology descriptions	4.4	Supported (proto)
KIP-1357 — Broker side custom assignors for streams groups	4.4	Supported (proto)

KIPs intentionally not implemented (with rationale):

• KIP-85, dynamic JAAS is a Java-specific auth framework; franz-go configures SASL mechanisms directly via the SASL() option
• KIP-92, per-partition consumer lag metrics; franz-go exposes per-batch information via hooks, and kadm.Lag handles lag monitoring via the admin API
• KIP-223, another JMX-style consumer metric; covered by hooks
• KIP-235, DNS alias support for secured connections; users can wire this up via a custom dialer and a custom kerberos SASL mechanism if needed
• KIP-359, adds CurrentLeaderEpoch to ProduceRequest so brokers reject writes to stale leaders; accepted but never shipped broker-side in any Kafka release
• KIP-421, variable substitution in config files; N/A, franz-go has no config file (same reason librdkafka skips it)
• KIP-436, start-time metric; covered by hooks
• KIP-517, more consumer poll-behavior metrics; covered by hooks
• KIP-896, which removes old client protocol API versions in Kafka 4.0; not relevant to clients, as clients do not control broker-supported protocol versions