Repository: twmb/franz-go
Pull request #467 (2023-06-02 14:58:02)

Serde Header
This PR introduces `SerdeHeader` and defines `ConfluentHeader` as the default implementation. The behavior of `Serde` is the same, ~one small but important difference is a fixed bug in the encoding of the message index (see the second part of the PR description)~. The changes are mostly just a refactoring and extraction of the header encoding/decoding logic. It makes `Serde` a bit more flexible and future-proof, since the header format can be switched to a different implementation.

The motivation behind this change is to expose the logic for encoding/decoding the Confluent wire format header. Using this change the caller can now decode the ID manually if needed and fetch the schema from the schema registry.

```go
val, err := serde.DecodeNew(b)
if errors.Is(err, sr.ErrNotRegistered) {
	id, _ := serde.Header().DecodeID(b)

	// fetch schema from schema registry using the ID...

	serde.Register(id, myVal, sr.DecodeFn(func(b []byte, v any) error {
		// decode using the schema ...
	})

	// retry decoding
	val, err = serde.DecodeNew(b)
}
```

The code above is just an illustration, I'm aware that the code would need to take care of request collapsing.

Related to the discussion in https://github.com/twmb/franz-go/issues/453.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2c674e76..ec1ccf4e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,63 @@
+v1.14.0
+===
+
+This release contains a few new APIs, one behavior change, and one minor bugfix.
+
+## Bugfixes
+
+Previously, `HookBrokerRead` and `HookBrokerE2E` could not be used at the same
+time. This has been fixed.
+
+## Behavior changes
+
+`PauseFetch{Topics,Partitions}` now causes the client to drop all buffered
+fetches and kill all in-flight fetch requests. Importantly, this also means
+that once you pause, it is no longer possible for what you paused to be
+returned while polling. Previously, the client made no attempt to clear
+internal buffers / in flight requests, meaning you could receive paused data
+for a while.
+
+Seed brokers now show up in logs as `seed_###` rather than `seed ###` (an
+underscore has been added).
+
+## Features
+
+* `kgo.Offset` now has an `EpochOffset` getter function that allows access
+  to the actual epoch and offset that are inside the opaque `Offset` type.
+* `AddConsumePartitions` allows adding individual partitions to consume, and
+  the new counterpart `RemoveConsumePartitions` allows removing individual
+  partitions from being consumed. Removing is different from purging, please
+  see the docs.
+* `KeepRetryableFetchErrors` bubbles up retryable errors to the end user that
+  are encountered while fetching. By default, these errors are stripped.
+* kversion now supports Kafka 3.5
+* kversion now supports version guessing against KRaft by default
+* `kgo.DialTLS` now exists to even more easily opt into TLS.
+* `kgo.Client.Opts` now exists to return the original options that were used
+  to configure the client, making initializing new clients easier.
+* `kgo.NodeName` returns a string form of a broker node name. Internally, seed
+  brokers use math.MinInt32 for node IDs, which shows up as massively negative
+  numbers in logs sometimes. `NodeName` can help convert that to `seed_<#>`.
+
+## Relevant commits
+
+- [`c3b083b`](https://github.com/twmb/franz-go/commit/c3b083b) **improvement** kgo: do not returned paused topics/partitions after pausing
+- [`e224e90`](https://github.com/twmb/franz-go/commit/e224e90) **bugfix** kgo: allow HookBrokerRead and HookBrokerE2E to both be called
+- [`875761a`](https://github.com/twmb/franz-go/commit/875761a) **feature** kgo Offset: add EpochOffset getter field
+- [`c5d0fc5`](https://github.com/twmb/franz-go/commit/c5d0fc5) kgo: add a debug log for stripping retryable errors from fetches
+- [`b45d663`](https://github.com/twmb/franz-go/commit/b45d663) kgo: add more context to opportunistic metadata loads while fetching
+- [`9dae366`](https://github.com/twmb/franz-go/commit/9dae366) kgo: allow retries on dial timeouts
+- [`00e4e76`](https://github.com/twmb/franz-go/commit/00e4e76) kgo: tolerate buggy v1 group member metadata
+- [`34c8b3d`](https://github.com/twmb/franz-go/commit/34c8b3d) **feature** kgo: add AddConsumePartitions and RemoveConsumePartitions
+- [`b5cafba`](https://github.com/twmb/franz-go/commit/b5cafba) sasl: validate non-empty user/pass/token
+- [`76d2e71`](https://github.com/twmb/franz-go/commit/76d2e71) **feature** kgo: add KeepRetryableFetchErrors
+- [`0df3ec0`](https://github.com/twmb/franz-go/commit/0df3ec0) kgo: fix new niche CI problem against Kafka 3.5
+- [`8ff1d0d`](https://github.com/twmb/franz-go/commit/8ff1d0d) **feature** pkg/kversion: attempt to guess KRaft by default as well
+- [`e92f5d9`](https://github.com/twmb/franz-go/commit/e92f5d9) **feature** pkg/kversion: detect v3.5
+- [`f1b923e`](https://github.com/twmb/franz-go/commit/f1b923e) **feature** kgo: add DialTLS option
+- [`9667967`](https://github.com/twmb/franz-go/commit/9667967) **feature** kgo.Client: add Opts to return the original opts, to allow seeding new clients
+- [`8e14928`](https://github.com/twmb/franz-go/commit/8e14928) **feature** kgo: add NodeName for easy user formatting of internal seed node IDs
+
 v1.13.6
 ===
 
diff --git a/README.md b/README.md
index eedefa78..0c264d33 100644
--- a/README.md
+++ b/README.md
@@ -340,6 +340,7 @@ generation.
 | [KIP-392](https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica) — Closest replica fetching w/ rack | 2.2 | Supported |
 | [KIP-394](https://cwiki.apache.org/confluence/display/KAFKA/KIP-394%3A+Require+member.id+for+initial+join+group+request) — Require member.id for initial join request |  2.2 | Supported |
 | [KIP-396](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=97551484) — Commit offsets manually | 2.4 | Supported |
+| [KIP-405](https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage) — Kafka Tiered Storage | 3.5 | Supported (protos) |
 | [KIP-412](https://cwiki.apache.org/confluence/display/KAFKA/KIP-412%3A+Extend+Admin+API+to+support+dynamic+application+log+levels) — Dynamic log levels w/ IncrementalAlterConfigs | 2.4 | Supported |
 | [KIP-429](https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol) — Incremental rebalance (see KAFKA-8179) | 2.4 | Supported |
 | [KIP-430](https://cwiki.apache.org/confluence/display/KAFKA/KIP-430+-+Return+Authorized+Operations+in+Describe+Responses) — Authorized ops in DescribeGroups | 2.3 | Supported |
@@ -397,6 +398,7 @@ generation.
 | [KIP-866](https://cwiki.apache.org/confluence/display/KAFKA/KIP-866+ZooKeeper+to+KRaft+Migration) — ZK to Raft RPC changes | 3.4 | Supported |
 | [KIP-893](https://cwiki.apache.org/confluence/display/KAFKA/KIP-893%3A+The+Kafka+protocol+should+support+nullable+structs) — Nullable structs in the protocol | 3.5 | Supported |
 | [KIP-899](https://cwiki.apache.org/confluence/display/KAFKA/KIP-899%3A+Allow+clients+to+rebootstrap) — Allow clients to rebootstrap | ? | Supported (`UpdateSeedBrokers`) |
+| [KIP-903](https://cwiki.apache.org/confluence/display/KAFKA/KIP-903%3A+Replicas+with+stale+broker+epoch+should+not+be+allowed+to+join+the+ISR) — Stale broker epoch fencing| 3.5 | Supported (proto) |
 
 Missing from above but included in librdkafka is:
 
diff --git a/generate/definitions/01_fetch b/generate/definitions/01_fetch
index 17a53ee8..5c25163d 100644
--- a/generate/definitions/01_fetch
+++ b/generate/definitions/01_fetch
@@ -10,14 +10,23 @@
 //
 // Starting in v13, topics must use UUIDs rather than their string name
 // identifiers.
-FetchRequest => key 1, max version 13, flexible v12+
+//
+// Version 15 adds the ReplicaState which includes new field ReplicaEpoch and
+// the ReplicaID, and deprecates the old ReplicaID (KIP-903).
+FetchRequest => key 1, max version 15, flexible v12+
   // The cluster ID, if known. This is used to validate metadata fetches
   // prior to broker registration.
   ClusterID: nullable-string(null) // tag 0
   // ReplicaID is the broker ID of performing the fetch request. Standard
   // clients should use -1. To be a "debug" replica, use -2. The debug
   // replica can be used to fetch messages from non-leaders.
-  ReplicaID: int32
+  ReplicaID: int32(-1) // v0-v14
+  // ReplicaState is a broker-only tag for v15+, see KIP-903 for more details.
+  ReplicaState: => // tag 1
+    // The replica ID of the follower, or -1 if this request is from a consumer.
+    ID: int32(-1)
+    // The epoch of this follower, or -1 if not available.
+    Epoch: int64(-1)
   // MaxWaitMillis is how long to wait for MinBytes to be hit before a broker
   // responds to a fetch request.
   MaxWaitMillis: int32
@@ -158,7 +167,11 @@ FetchResponse =>
       // OFFSET_OUT_OF_RANGE is returned if requesting an offset past the
       // current end offset or before the beginning offset.
       //
-      // UNKNOWN_TOPIC_ID is returned if using uuid's and the uuid is unknown.
+      // UNKNOWN_TOPIC_ID is returned if using uuid's and the uuid is unknown
+      // (v13+ / Kafka 3.1+).
+      //
+      // OFFSET_MOVED_TO_TIERED_STORAGE is returned if a follower is trying to
+      // fetch from an offset that is now in tiered storage.
       ErrorCode: int16
       // HighWatermark is the current high watermark for this partition,
       // that is, the current offset that is on all in sync replicas.
diff --git a/generate/definitions/02_list_offsets b/generate/definitions/02_list_offsets
index f6212eb2..113f4c7d 100644
--- a/generate/definitions/02_list_offsets
+++ b/generate/definitions/02_list_offsets
@@ -7,7 +7,10 @@
 //
 // Version 7, introduced in Kafka 3.0, supports -3 as a timestamp to return
 // the timestamp and offset for the record with the largest timestamp.
-ListOffsetsRequest => key 2, max version 7, flexible v6+
+//
+// Version 8, introduced in Kafka 3.4, supports -4 as a timestamp to return
+// the local log start offset (in the context of tiered storage, see KIP-405).
+ListOffsetsRequest => key 2, max version 8, flexible v6+
   // ReplicaID is the broker ID to get offsets from. As a Kafka client, use -1.
   // The consumer replica ID (-1) causes requests to only succeed if issued
   // against the leader broker.
@@ -53,6 +56,11 @@ ListOffsetsRequest => key 2, max version 7, flexible v6+
       // to determine topic "liveness" (when was the last produce?).
       // Previously, this was not easy to determine. See KIP-734 for more
       // detail.
+      //
+      // If you are talking to Kafka 3.4+ and using request version 8+ (for
+      // KIP-405), the new special timestamp -4 returns the local log start
+      // offset. In the context of tiered storage, the earliest local log start
+      // offset is the offset actually available on disk on the broker.
       Timestamp: int64
       // MaxNumOffsets is the maximum number of offsets to report.
       // This was removed after v0.
diff --git a/generate/definitions/56_alter_partition b/generate/definitions/56_alter_partition
index e331f312..872ac8bd 100644
--- a/generate/definitions/56_alter_partition
+++ b/generate/definitions/56_alter_partition
@@ -1,6 +1,8 @@
 // AlterPartitionRequest, proposed in KIP-497 and introduced in Kafka 2.7.0,
 // is an admin request to modify ISR.
-AlterPartitionRequest => key 56, max version 2, flexible v0+, admin
+//
+// Version 3 was added for KIP-903 and replaced NewISR.
+AlterPartitionRequest => key 56, max version 3, flexible v0+, admin
   // The ID of the requesting broker.
   BrokerID: int32
   // The epoch of the requesting broker.
@@ -13,7 +15,12 @@ AlterPartitionRequest => key 56, max version 2, flexible v0+, admin
       // The leader epoch of this partition.
       LeaderEpoch: int32
       // The ISR for this partition.
-      NewISR: [int32]
+      NewISR: [int32] // v0-v2
+      NewEpochISR: [=>] // v3+
+        // The broker ID .
+        BrokerID: int32
+        // The broker's epoch; -1 if the epoch check is not supported.
+        BrokerEpoch: int32(-1)
       // 1 if the partition is recovering from unclean leader election; 0 otherwise
       LeaderRecoveryState: int8 // v1+
       // The expected epoch of the partition which is being updated.
diff --git a/go.mod b/go.mod
index ffcb4a8d..d945fef6 100644
--- a/go.mod
+++ b/go.mod
@@ -3,10 +3,10 @@ module github.com/twmb/franz-go
 go 1.18
 
 require (
-	github.com/klauspost/compress v1.16.3
-	github.com/pierrec/lz4/v4 v4.1.17
-	github.com/twmb/franz-go/pkg/kmsg v1.4.0
-	golang.org/x/crypto v0.7.0
+	github.com/klauspost/compress v1.16.7
+	github.com/pierrec/lz4/v4 v4.1.18
+	github.com/twmb/franz-go/pkg/kmsg v1.6.1
+	golang.org/x/crypto v0.11.0
 )
 
 retract v1.11.4 // This version is actually a breaking change and requires a major version change.
diff --git a/go.sum b/go.sum
index fd07af94..02d6ebc1 100644
--- a/go.sum
+++ b/go.sum
@@ -1,8 +1,8 @@
-github.com/klauspost/compress v1.16.3 h1:XuJt9zzcnaz6a16/OU53ZjWp/v7/42WcR5t2a0PcNQY=
-github.com/klauspost/compress v1.16.3/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
-github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
-github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
-github.com/twmb/franz-go/pkg/kmsg v1.4.0 h1:tbp9hxU6m8qZhQTlpGiaIJOm4BXix5lsuEZ7K00dF0s=
-github.com/twmb/franz-go/pkg/kmsg v1.4.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=
-golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A=
-golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU=
+github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I=
+github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
+github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/twmb/franz-go/pkg/kmsg v1.6.1 h1:tm6hXPv5antMHLasTfKv9R+X03AjHSkSkXhQo2c5ALM=
+github.com/twmb/franz-go/pkg/kmsg v1.6.1/go.mod h1:se9Mjdt0Nwzc9lnjJ0HyDtLyBnaBDAd7pCje47OhSyw=
+golang.org/x/crypto v0.11.0 h1:6Ewdq3tDic1mg5xRO4milcWCfMVQhI4NkqWWvqejpuA=
+golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
diff --git a/pkg/kadm/metadata.go b/pkg/kadm/metadata.go
index 1b69f1fb..b6f9c8af 100644
--- a/pkg/kadm/metadata.go
+++ b/pkg/kadm/metadata.go
@@ -234,12 +234,14 @@ func (cl *Client) metadata(ctx context.Context, noTopics bool, topics []string)
 			return Metadata{}, err
 		}
 		td := TopicDetail{
-			Topic:      *t.Topic,
 			ID:         t.TopicID,
 			Partitions: make(map[int32]PartitionDetail),
 			IsInternal: t.IsInternal,
 			Err:        kerr.ErrorForCode(t.ErrorCode),
 		}
+		if t.Topic != nil {
+			td.Topic = *t.Topic
+		}
 		for _, p := range t.Partitions {
 			td.Partitions[p.Partition] = PartitionDetail{
 				Topic:     td.Topic,
@@ -403,6 +405,11 @@ func (cl *Client) listOffsets(ctx context.Context, isolation int8, timestamp int
 	if err != nil {
 		return nil, err
 	}
+	for _, td := range tds {
+		if td.Err != nil {
+			return nil, td.Err
+		}
+	}
 
 	// If we request with timestamps, we may request twice: once for after
 	// timestamps, and once for any -1 (and no error) offsets where the
diff --git a/pkg/kerr/kerr.go b/pkg/kerr/kerr.go
index e05e3188..0d837d50 100644
--- a/pkg/kerr/kerr.go
+++ b/pkg/kerr/kerr.go
@@ -182,6 +182,7 @@ var (
 	FetchSessionTopicIDError           = &Error{"FETCH_SESSION_TOPIC_ID_ERROR", 106, true, "The fetch session encountered inconsistent topic ID usage."}
 	IneligibleReplica                  = &Error{"INELIGIBLE_REPLICA", 107, false, "The new ISR contains at least one ineligible replica."}
 	NewLeaderElected                   = &Error{"NEW_LEADER_ELECTED", 108, false, "The AlterPartition request successfully updated the partition state but the leader has changed."}
+	OffsetMovedToTieredStorage         = &Error{"OFFSET_MOVED_TO_TIERED_STORAGE", 109, false, "The requested offset is moved to tiered storage."}
 )
 
 var code2err = map[int16]error{
@@ -295,4 +296,5 @@ var code2err = map[int16]error{
 	106: FetchSessionTopicIDError,
 	107: IneligibleReplica,
 	108: NewLeaderElected,
+	109: OffsetMovedToTieredStorage,
 }
diff --git a/pkg/kgo/broker.go b/pkg/kgo/broker.go
index 9bad4d4a..5383e7b4 100644
--- a/pkg/kgo/broker.go
+++ b/pkg/kgo/broker.go
@@ -82,11 +82,22 @@ type promisedResp struct {
 	readEnqueue  time.Time
 }
 
+// NodeName returns the name of a node, given the kgo internal node ID.
+//
+// Internally, seed brokers are stored with very negative node IDs, and these
+// node IDs are visible in the BrokerMetadata struct. You can use NodeName to
+// convert the negative node ID into "seed_#". Brokers discovered through
+// metadata responses have standard non-negative numbers and this function just
+// returns the number as a string.
+func NodeName(nodeID int32) string {
+	return logID(nodeID)
+}
+
 func logID(id int32) string {
 	if id >= -10 {
 		return strconv.FormatInt(int64(id), 10)
 	}
-	return "seed " + strconv.FormatInt(int64(id)-math.MinInt32, 10)
+	return "seed_" + strconv.FormatInt(int64(id)-math.MinInt32, 10)
 }
 
 // BrokerMetadata is metadata for a broker.
@@ -96,7 +107,8 @@ type BrokerMetadata struct {
 	// NodeID is the broker node ID.
 	//
 	// Seed brokers will have very negative IDs; kgo does not try to map
-	// seed brokers to loaded brokers.
+	// seed brokers to loaded brokers. You can use NodeName to convert
+	// the seed node ID into a formatted string.
 	NodeID int32
 
 	// Port is the port of the broker.
@@ -1205,10 +1217,10 @@ func (cxn *brokerCxn) readResponse(
 	bytesRead, buf, readWait, timeToRead, readErr := cxn.readConn(ctx, timeout, readEnqueue)
 
 	cxn.cl.cfg.hooks.each(func(h Hook) {
-		switch h := h.(type) {
-		case HookBrokerRead:
+		if h, ok := h.(HookBrokerRead); ok {
 			h.OnBrokerRead(cxn.b.meta, key, bytesRead, readWait, timeToRead, readErr)
-		case HookBrokerE2E:
+		}
+		if h, ok := h.(HookBrokerE2E); ok {
 			h.OnBrokerE2E(cxn.b.meta, key, BrokerE2E{
 				BytesWritten: bytesWritten,
 				BytesRead:    bytesRead,
diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go
index b421932b..b4cf0e3e 100644
--- a/pkg/kgo/client.go
+++ b/pkg/kgo/client.go
@@ -33,7 +33,8 @@ var crc32c = crc32.MakeTable(crc32.Castagnoli) // record crc's use Castagnoli ta
 
 // Client issues requests and handles responses to a Kafka cluster.
 type Client struct {
-	cfg cfg
+	cfg  cfg
+	opts []Opt
 
 	ctx       context.Context
 	ctxCancel func()
@@ -248,6 +249,8 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.dialFn}
 	case namefn(DialTLSConfig):
 		return []any{cfg.dialTLS}
+	case namefn(DialTLS):
+		return []any{cfg.dialTLS != nil}
 	case namefn(SeedBrokers):
 		return []any{cfg.seedBrokers}
 	case namefn(MaxVersions):
@@ -347,6 +350,8 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.maxConcurrentFetches}
 	case namefn(Rack):
 		return []any{cfg.rack}
+	case namefn(KeepRetryableFetchErrors):
+		return []any{cfg.keepRetryableFetchErrors}
 
 	case namefn(AdjustFetchOffsetsFn):
 		return []any{cfg.adjustOffsetsBeforeAssign}
@@ -449,6 +454,7 @@ func NewClient(opts ...Opt) (*Client, error) {
 
 	cl := &Client{
 		cfg:       cfg,
+		opts:      opts,
 		ctx:       ctx,
 		ctxCancel: cancel,
 
@@ -511,6 +517,14 @@ func NewClient(opts ...Opt) (*Client, error) {
 	return cl, nil
 }
 
+// Opts returns the options that were used to create this client. This can be
+// as a base to generate a new client, where you can add override options to
+// the end of the original input list. If you want to know a specific option
+// value, you can use ConfigValue or ConfigValues.
+func (cl *Client) Opts() []Opt {
+	return cl.opts
+}
+
 func (cl *Client) loadSeeds() []*broker {
 	return cl.seeds.Load().([]*broker)
 }
@@ -1108,7 +1122,7 @@ type failDial struct{ fails int8 }
 // repeatedly fails, we need to forget our cache to force a re-load: the broker
 // may have completely died.
 func (d *failDial) isRepeatedDialFail(err error) bool {
-	if isDialErr(err) {
+	if isAnyDialErr(err) {
 		d.fails++
 		if d.fails == 3 {
 			d.fails = 0
diff --git a/pkg/kgo/config.go b/pkg/kgo/config.go
index b9a5128a..011d347a 100644
--- a/pkg/kgo/config.go
+++ b/pkg/kgo/config.go
@@ -143,7 +143,7 @@ type cfg struct {
 
 	maxConcurrentFetches     int
 	disableFetchSessions     bool
-	keepFetchRetryableErrors bool
+	keepRetryableFetchErrors bool
 
 	topics     map[string]*regexp.Regexp   // topics to consume; if regex is true, values are compiled regular expressions
 	partitions map[string]map[int32]Offset // partitions to directly consume from
@@ -642,6 +642,12 @@ func DialTLSConfig(c *tls.Config) Opt {
 	return clientOpt{func(cfg *cfg) { cfg.dialTLS = c }}
 }
 
+// DialTLS opts into dialing brokers with TLS. This is a shortcut for
+// DialTLSConfig with an empty config. See DialTLSConfig for more details.
+func DialTLS() Opt {
+	return DialTLSConfig(new(tls.Config))
+}
+
 // SeedBrokers sets the seed brokers for the client to use, overriding the
 // default 127.0.0.1:9092.
 //
@@ -1343,7 +1349,7 @@ func ConsumePreferringLagFn(fn PreferLagFn) ConsumerOpt {
 	return consumerOpt{func(cfg *cfg) { cfg.preferLagFn = fn }}
 }
 
-// KeepFetchRetryableErrors switches the client to always return any retryable
+// KeepRetryableFetchErrors switches the client to always return any retryable
 // broker error when fetching, rather than stripping them. By default, the
 // client strips retryable errors from fetch responses; these are usually
 // signals that a client needs to update its metadata to learn of where a
@@ -1353,10 +1359,8 @@ func ConsumePreferringLagFn(fn PreferLagFn) ConsumerOpt {
 // events. For example, if you want to react to you yourself deleting a topic,
 // you can watch for either UNKNOWN_TOPIC_OR_PARTITION or UNKNOWN_TOPIC_ID
 // errors being returned in fetches (and ignore the other errors).
-//
-// TODO not exported / usable yet
-func keepFetchRetryableErrors() ConsumerOpt {
-	return consumerOpt{func(cfg *cfg) { cfg.keepFetchRetryableErrors = true }}
+func KeepRetryableFetchErrors() ConsumerOpt {
+	return consumerOpt{func(cfg *cfg) { cfg.keepRetryableFetchErrors = true }}
 }
 
 //////////////////////////////////
diff --git a/pkg/kgo/consumer.go b/pkg/kgo/consumer.go
index 9dde7c0c..ddeb81a2 100644
--- a/pkg/kgo/consumer.go
+++ b/pkg/kgo/consumer.go
@@ -46,6 +46,15 @@ func (o Offset) String() string {
 	}
 }
 
+// EpochOffset returns this offset as an EpochOffset, allowing visibility into
+// what this offset actually currently is.
+func (o Offset) EpochOffset() EpochOffset {
+	return EpochOffset{
+		Epoch:  o.epoch,
+		Offset: o.at,
+	}
+}
+
 // NewOffset creates and returns an offset to use in ConsumePartitions or
 // ConsumeResetOffset.
 //
@@ -536,11 +545,9 @@ func (cl *Client) UpdateFetchMaxBytes(maxBytes, maxPartBytes int32) {
 // PauseFetchTopics sets the client to no longer fetch the given topics and
 // returns all currently paused topics. Paused topics persist until resumed.
 // You can call this function with no topics to simply receive the list of
-// currently paused topics.
-//
-// In contrast to the canonical Java client, this function does not clear
-// anything currently buffered. Buffered fetches containing paused topics are
-// still returned from polling.
+// currently paused topics. Pausing topics drops everything currently buffered
+// and kills any in flight fetch requests to ensure nothing that is paused
+// can be returned anymore from polling.
 //
 // Pausing topics is independent from pausing individual partitions with the
 // PauseFetchPartitions method. If you pause partitions for a topic with
@@ -555,6 +562,7 @@ func (cl *Client) PauseFetchTopics(topics ...string) []string {
 
 	c.pausedMu.Lock()
 	defer c.pausedMu.Unlock()
+	defer c.assignPartitions(nil, assignBumpSession, nil, fmt.Sprintf("pausing fetch topics %v", topics))
 
 	paused := c.clonePaused()
 	paused.addTopics(topics...)
@@ -565,11 +573,9 @@ func (cl *Client) PauseFetchTopics(topics ...string) []string {
 // PauseFetchPartitions sets the client to no longer fetch the given partitions
 // and returns all currently paused partitions. Paused partitions persist until
 // resumed. You can call this function with no partitions to simply receive the
-// list of currently paused partitions.
-//
-// In contrast to the canonical Java client, this function does not clear
-// anything currently buffered. Buffered fetches containing paused partitions
-// are still returned from polling.
+// list of currently paused partitions. Pausing partitions drops everything
+// currently buffered and kills any in flight fetch requests to ensure nothing
+// that is paused can be returned anymore from polling.
 //
 // Pausing individual partitions is independent from pausing topics with the
 // PauseFetchTopics method. If you pause partitions for a topic with
@@ -584,6 +590,7 @@ func (cl *Client) PauseFetchPartitions(topicPartitions map[string][]int32) map[s
 
 	c.pausedMu.Lock()
 	defer c.pausedMu.Unlock()
+	defer c.assignPartitions(nil, assignBumpSession, nil, fmt.Sprintf("pausing fetch partitions %v", topicPartitions))
 
 	paused := c.clonePaused()
 	paused.addPartitions(topicPartitions)
@@ -749,6 +756,91 @@ func (cl *Client) AddConsumeTopics(topics ...string) {
 	cl.triggerUpdateMetadataNow("from AddConsumeTopics")
 }
 
+// AddConsumePartitions adds new partitions to be consumed at the given
+// offsets. This function works only for direct, non-regex consumers.
+func (cl *Client) AddConsumePartitions(partitions map[string]map[int32]Offset) {
+	c := &cl.consumer
+	if c.d == nil || cl.cfg.regex {
+		return
+	}
+	var topics []string
+	for t, ps := range partitions {
+		if len(ps) == 0 {
+			delete(partitions, t)
+			continue
+		}
+		topics = append(topics, t)
+	}
+	if len(partitions) == 0 {
+		return
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.d.tps.storeTopics(topics)
+	for t, ps := range partitions {
+		if c.d.ps[t] == nil {
+			c.d.ps[t] = make(map[int32]Offset)
+		}
+		for p, o := range ps {
+			c.d.m.add(t, p)
+			c.d.ps[t][p] = o
+		}
+	}
+	cl.triggerUpdateMetadataNow("from AddConsumePartitions")
+}
+
+// RemoveConsumePartitions removes partitions from being consumed. This
+// function works only for direct, non-regex consumers.
+//
+// This method does not purge the concept of any topics from the client -- if
+// you remove all partitions from a topic that was being consumed, metadata
+// fetches will still occur for the topic. If you want to remove the topic
+// entirely, use PurgeTopicsFromClient.
+//
+// If you specified ConsumeTopics and this function removes all partitions for
+// a topic, the topic will no longer be consumed.
+func (cl *Client) RemoveConsumePartitions(partitions map[string][]int32) {
+	c := &cl.consumer
+	if c.d == nil || cl.cfg.regex {
+		return
+	}
+	for t, ps := range partitions {
+		if len(ps) == 0 {
+			delete(partitions, t)
+			continue
+		}
+	}
+	if len(partitions) == 0 {
+		return
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	removeOffsets := make(map[string]map[int32]Offset, len(partitions))
+	for t, ps := range partitions {
+		removePartitionOffsets := make(map[int32]Offset, len(ps))
+		for _, p := range ps {
+			removePartitionOffsets[p] = Offset{}
+		}
+		removeOffsets[t] = removePartitionOffsets
+	}
+
+	c.assignPartitions(removeOffsets, assignInvalidateMatching, c.d.tps, fmt.Sprintf("remove of %v requested", partitions))
+	for t, ps := range partitions {
+		for _, p := range ps {
+			c.d.using.remove(t, p)
+			c.d.m.remove(t, p)
+			delete(c.d.ps[t], p)
+		}
+		if len(c.d.ps[t]) == 0 {
+			delete(c.d.ps, t)
+		}
+	}
+}
+
 // assignHow controls how assignPartitions operates.
 type assignHow int8
 
@@ -774,6 +866,10 @@ const (
 	// The counterpart to assignInvalidateMatching, assignSetMatching
 	// resets all matching partitions to the specified offset / epoch.
 	assignSetMatching
+
+	// For pausing, we want to drop anything inflight. We start a new
+	// session with the old tps.
+	assignBumpSession
 )
 
 func (h assignHow) String() string {
@@ -788,6 +884,8 @@ func (h assignHow) String() string {
 		return "unassigning and purging any partition matching the input topics"
 	case assignSetMatching:
 		return "reassigning any currently assigned matching partition to the input"
+	case assignBumpSession:
+		return "bumping internal consumer session to drop anything currently in flight"
 	}
 	return ""
 }
@@ -858,6 +956,8 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
 		// if we had no session before, which is why we need to pass in
 		// our topicPartitions.
 		session = c.guardSessionChange(tps)
+	} else if how == assignBumpSession {
+		loadOffsets, tps = c.stopSession()
 	} else {
 		loadOffsets, _ = c.stopSession()
 
@@ -904,7 +1004,7 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
 		// assignment went straight to listing / epoch loading, and
 		// that list/epoch never finished.
 		switch how {
-		case assignWithoutInvalidating:
+		case assignWithoutInvalidating, assignBumpSession:
 			// Nothing to do -- this is handled above.
 		case assignInvalidateAll:
 			loadOffsets = listOrEpochLoads{}
@@ -945,7 +1045,7 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
 
 	// This assignment could contain nothing (for the purposes of
 	// invalidating active fetches), so we only do this if needed.
-	if len(assignments) == 0 || how == assignInvalidateMatching || how == assignPurgeMatching || how == assignSetMatching {
+	if len(assignments) == 0 || how == assignInvalidateMatching || how == assignPurgeMatching || how == assignSetMatching || how == assignBumpSession {
 		return
 	}
 
@@ -1695,7 +1795,7 @@ func (s *consumerSession) handleListOrEpochResults(loaded loadedOffsets) (reload
 
 		default: // from ErrorCode in a response, or broker request err, or request is canceled as our session is ending
 			reloads.addLoad(load.topic, load.partition, loaded.loadType, load.request)
-			if !kerr.IsRetriable(load.err) && !isRetryableBrokerErr(load.err) && !isDialErr(load.err) && !isContextErr(load.err) { // non-retryable response error; signal such in a response
+			if !kerr.IsRetriable(load.err) && !isRetryableBrokerErr(load.err) && !isDialNonTimeoutErr(load.err) && !isContextErr(load.err) { // non-retryable response error; signal such in a response
 				s.c.addFakeReadyForDraining(load.topic, load.partition, load.err, fmt.Sprintf("notification of non-retryable error from %s request", loaded.loadType))
 			}
 
diff --git a/pkg/kgo/consumer_direct.go b/pkg/kgo/consumer_direct.go
index 41869596..1d328c60 100644
--- a/pkg/kgo/consumer_direct.go
+++ b/pkg/kgo/consumer_direct.go
@@ -2,10 +2,11 @@ package kgo
 
 type directConsumer struct {
 	cfg    *cfg
-	tps    *topicsPartitions // data for topics that the user assigned
-	using  mtmps             // topics we are currently using
-	m      mtmps             // mirrors cfg.topics and cfg.partitions, but can change with Purge or Add
-	reSeen map[string]bool   // topics we evaluated against regex, and whether we want them or not
+	tps    *topicsPartitions           // data for topics that the user assigned
+	using  mtmps                       // topics we are currently using
+	m      mtmps                       // mirrors cfg.topics and cfg.partitions, but can change with Purge or Add
+	ps     map[string]map[int32]Offset // mirrors cfg.partitions, changed in Purge or Add
+	reSeen map[string]bool             // topics we evaluated against regex, and whether we want them or not
 }
 
 func (c *consumer) initDirect() {
@@ -15,6 +16,7 @@ func (c *consumer) initDirect() {
 		reSeen: make(map[string]bool),
 		using:  make(mtmps),
 		m:      make(mtmps),
+		ps:     make(map[string]map[int32]Offset),
 	}
 	c.d = d
 
@@ -28,6 +30,11 @@ func (c *consumer) initDirect() {
 		for partition := range partitions {
 			d.m.add(topic, partition)
 		}
+		p := make(map[int32]Offset, len(partitions))
+		for partition, offset := range partitions {
+			p[partition] = offset
+		}
+		d.ps[topic] = p
 	}
 	for topic := range d.cfg.topics {
 		topics = append(topics, topic)
@@ -106,7 +113,7 @@ func (d *directConsumer) findNewAssignments() map[string]map[int32]Offset {
 		// we set those. We only use partitions from topics that have
 		// not been purged.
 		for topic := range d.m {
-			for partition, offset := range d.cfg.partitions[topic] {
+			for partition, offset := range d.ps[topic] {
 				toUseTopic, exists := toUse[topic]
 				if !exists {
 					toUseTopic = make(map[int32]Offset, 10)
diff --git a/pkg/kgo/errors.go b/pkg/kgo/errors.go
index c20d516a..8d574b90 100644
--- a/pkg/kgo/errors.go
+++ b/pkg/kgo/errors.go
@@ -44,7 +44,7 @@ func isRetryableBrokerErr(err error) bool {
 		// If a dial fails, potentially we could retry if the resolver
 		// had a temporary hiccup, but we will err on the side of this
 		// being a slightly less temporary error.
-		return !isDialErr(err)
+		return !isDialNonTimeoutErr(err)
 	}
 	// EOF can be returned if a broker kills a connection unexpectedly, and
 	// we can retry that. Same for ErrClosed.
@@ -88,7 +88,12 @@ func isRetryableBrokerErr(err error) bool {
 	return false
 }
 
-func isDialErr(err error) bool {
+func isDialNonTimeoutErr(err error) bool {
+	var ne *net.OpError
+	return errors.As(err, &ne) && ne.Op == "dial" && !ne.Timeout()
+}
+
+func isAnyDialErr(err error) bool {
 	var ne *net.OpError
 	return errors.As(err, &ne) && ne.Op == "dial"
 }
diff --git a/pkg/kgo/group_balancer.go b/pkg/kgo/group_balancer.go
index a3f21017..87a3a248 100644
--- a/pkg/kgo/group_balancer.go
+++ b/pkg/kgo/group_balancer.go
@@ -204,8 +204,23 @@ func NewConsumerBalancer(balance ConsumerBalancerBalance, members []kmsg.JoinGro
 	for i, member := range members {
 		meta := &b.metadatas[i]
 		meta.Default()
-		if err := meta.ReadFrom(member.ProtocolMetadata); err != nil {
-			return nil, fmt.Errorf("unable to read member metadata: %v", err)
+		memberMeta := member.ProtocolMetadata
+		if err := meta.ReadFrom(memberMeta); err != nil {
+			// Some buggy clients claimed support for v1 but then
+			// did not add OwnedPartitions, resulting in a short
+			// metadata. If we fail at reading and the version is
+			// v1, we retry again as v0. We do not support other
+			// versions because hopefully other clients stop
+			// claiming higher and higher version support and not
+			// actually supporting them. Sarama has a similarish
+			// workaround. See #493.
+			if bytes.HasPrefix(memberMeta, []byte{0, 1}) {
+				memberMeta[0] = 0
+				memberMeta[0] = 0
+				if err = meta.ReadFrom(memberMeta); err != nil {
+					return nil, fmt.Errorf("unable to read member metadata: %v", err)
+				}
+			}
 		}
 		for _, topic := range meta.Topics {
 			b.topics[topic] = struct{}{}
diff --git a/pkg/kgo/metadata.go b/pkg/kgo/metadata.go
index 29042eb1..6007b097 100644
--- a/pkg/kgo/metadata.go
+++ b/pkg/kgo/metadata.go
@@ -287,6 +287,8 @@ loop:
 	}
 }
 
+var errMissingTopic = errors.New("topic_missing")
+
 // Updates all producer and consumer partition data, returning whether a new
 // update needs scheduling or if an error occurred.
 //
@@ -339,6 +341,8 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 	}
 	groupExternal.updateLatest(latest)
 
+	const maxMissTime = 15 * time.Second
+
 	// If we are consuming with regex and fetched all topics, the metadata
 	// may have returned topics the consumer is not yet tracking. We ensure
 	// that we will store the topics at the end of our metadata update.
@@ -352,12 +356,20 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 		defer tpsConsumer.storeData(tpsConsumerLoad)
 
 		// For regex consuming, if a topic is not returned in the
-		// response that we were previously consuming, we assume the
-		// topic has been deleted and purge it.
+		// response and for at least maxMissTime from when we first
+		// discovered it, we assume the topic has been deleted and
+		// purge it. We allow for maxMissTime because (in testing
+		// locally) Kafka can originally broadcast a newly created
+		// topic exists and then fail to broadcast that info again for
+		// a while.
 		var purgeTopics []string
-		for topic := range tpsConsumerLoad {
+		for topic, tps := range tpsConsumerLoad {
 			if _, ok := latest[topic]; !ok {
-				purgeTopics = append(purgeTopics, topic)
+				if td := tps.load(); td.when != 0 && time.Since(time.Unix(td.when, 0)) > maxMissTime {
+					purgeTopics = append(purgeTopics, td.topic)
+				} else {
+					retryWhy.add(topic, -1, errMissingTopic)
+				}
 			}
 		}
 		if len(purgeTopics) > 0 {
@@ -395,7 +407,7 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 		}
 	}()
 
-	var missingProduceTopics []string
+	var missingProduceTopics []*topicPartitions
 	for _, m := range []struct {
 		priors    map[string]*topicPartitions
 		isProduce bool
@@ -407,7 +419,7 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 			newParts, exists := latest[topic]
 			if !exists {
 				if m.isProduce {
-					missingProduceTopics = append(missingProduceTopics, topic)
+					missingProduceTopics = append(missingProduceTopics, priorParts)
 				}
 				continue
 			}
@@ -422,12 +434,33 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 			)
 		}
 	}
+
+	// For all produce topics that were missing, we want to bump their
+	// retries that a failure happened. However, if we are regex consuming,
+	// then it is possible in a rare scenario for the broker to not return
+	// a topic that actually does exist and that we previously received a
+	// metadata response for. This is handled above for consuming, we now
+	// handle it the same way for consuming.
 	if len(missingProduceTopics) > 0 {
-		cl.bumpMetadataFailForTopics(
-			tpsProducerLoad,
-			errors.New("metadata request did not return this topic"),
-			missingProduceTopics...,
-		)
+		var bumpFail []string
+		for _, tps := range missingProduceTopics {
+			if all {
+				if td := tps.load(); td.when != 0 && time.Since(time.Unix(td.when, 0)) > maxMissTime {
+					bumpFail = append(bumpFail, td.topic)
+				} else {
+					retryWhy.add(td.topic, -1, errMissingTopic)
+				}
+			} else {
+				bumpFail = append(bumpFail, tps.load().topic)
+			}
+		}
+		if len(bumpFail) > 0 {
+			cl.bumpMetadataFailForTopics(
+				tpsProducerLoad,
+				fmt.Errorf("metadata request did not return topics: %v", bumpFail),
+				bumpFail...,
+			)
+		}
 	}
 
 	return retryWhy, nil
@@ -444,6 +477,7 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 type metadataTopic struct {
 	loadErr    error
 	isInternal bool
+	topic      string
 	partitions []metadataPartition
 }
 
@@ -454,6 +488,8 @@ func (mt *metadataTopic) newPartitions(cl *Client, isProduce bool) *topicPartiti
 		isInternal:         mt.isInternal,
 		partitions:         make([]*topicPartition, 0, n),
 		writablePartitions: make([]*topicPartition, 0, n),
+		topic:              mt.topic,
+		when:               time.Now().Unix(),
 	}
 	for i := range mt.partitions {
 		p := mt.partitions[i].newPartition(cl, isProduce)
@@ -539,6 +575,7 @@ func (cl *Client) fetchTopicMetadata(all bool, reqTopics []string) (map[string]*
 		mt := &metadataTopic{
 			loadErr:    kerr.ErrorForCode(topicMeta.ErrorCode),
 			isInternal: topicMeta.IsInternal,
+			topic:      topic,
 			partitions: make([]metadataPartition, 0, len(topicMeta.Partitions)),
 		}
 
@@ -646,6 +683,9 @@ func (cl *Client) mergeTopicPartitions(
 
 	lv.loadErr = r.loadErr
 	lv.isInternal = r.isInternal
+	if lv.when == 0 {
+		lv.when = r.when
+	}
 
 	// If the load had an error for the entire topic, we set the load error
 	// but keep our stale partition information. For anything being
diff --git a/pkg/kgo/sink.go b/pkg/kgo/sink.go
index ea013efc..3a1ef694 100644
--- a/pkg/kgo/sink.go
+++ b/pkg/kgo/sink.go
@@ -326,7 +326,7 @@ func (s *sink) produce(sem <-chan struct{}) bool {
 		batchesStripped, err := s.doTxnReq(req, txnReq)
 		if err != nil {
 			switch {
-			case isRetryableBrokerErr(err) || isDialErr(err):
+			case isRetryableBrokerErr(err) || isDialNonTimeoutErr(err):
 				s.cl.bumpRepeatedLoadErr(err)
 				s.cl.cfg.logger.Log(LogLevelWarn, "unable to AddPartitionsToTxn due to retryable broker err, bumping client's buffered record load errors by 1 and retrying", "err", err)
 				s.cl.triggerUpdateMetadata(false, "attempting to refresh broker list due to failed AddPartitionsToTxn requests")
@@ -532,7 +532,7 @@ func (s *sink) handleReqClientErr(req *produceRequest, err error) {
 		fallthrough
 
 	case errors.Is(err, errUnknownBroker),
-		isDialErr(err),
+		isDialNonTimeoutErr(err),
 		isRetryableBrokerErr(err):
 		updateMeta := !isRetryableBrokerErr(err)
 		if updateMeta {
@@ -1269,7 +1269,7 @@ func (recBuf *recBuf) bumpRepeatedLoadErr(err error) {
 	var (
 		canFail        = !recBuf.cl.idempotent() || batch0.canFailFromLoadErrs // we can only fail if we are not idempotent or if we have no outstanding requests
 		batch0Fail     = batch0.maybeFailErr(&recBuf.cl.cfg) != nil            // timeout, retries, or aborting
-		netErr         = isRetryableBrokerErr(err) || isDialErr(err)           // we can fail if this is *not* a network error
+		netErr         = isRetryableBrokerErr(err) || isDialNonTimeoutErr(err) // we can fail if this is *not* a network error
 		retryableKerr  = kerr.IsRetriable(err)                                 // we fail if this is not a retryable kerr,
 		isUnknownLimit = recBuf.checkUnknownFailLimit(err)                     // or if it is, but it is UnknownTopicOrPartition and we are at our limit
 
diff --git a/pkg/kgo/source.go b/pkg/kgo/source.go
index ed5ded6d..d90cd4d5 100644
--- a/pkg/kgo/source.go
+++ b/pkg/kgo/source.go
@@ -624,7 +624,7 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 	}
 
 	var didBackoff bool
-	backoff := func() {
+	backoff := func(why interface{}) {
 		// We preemptively allow more fetches (since we are not buffering)
 		// and reset our session because of the error (who knows if kafka
 		// processed the request but the client failed to receive it).
@@ -633,7 +633,7 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 		s.session.reset()
 		didBackoff = true
 
-		s.cl.triggerUpdateMetadata(false, "opportunistic load during source backoff") // as good a time as any
+		s.cl.triggerUpdateMetadata(false, fmt.Sprintf("opportunistic load during source backoff: %v", why)) // as good a time as any
 		s.consecutiveFailures++
 		after := time.NewTimer(s.cl.cfg.retryBackoff(s.consecutiveFailures))
 		defer after.Stop()
@@ -652,7 +652,7 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 	// but that is fine; we may just re-request too early and fall into
 	// another backoff.
 	if err != nil {
-		backoff()
+		backoff(err)
 		return
 	}
 
@@ -774,7 +774,7 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 		// fetching from topics that were deleted. We want to back off
 		// a bit rather than spin-loop immediately re-requesting
 		// deleted topics.
-		backoff()
+		backoff("empty fetch response due to all partitions having retryable errors")
 	}
 	return
 }
@@ -796,11 +796,20 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 ) {
 	f = Fetch{Topics: make([]FetchTopic, 0, len(resp.Topics))}
 	var (
-		updateWhy       multiUpdateWhy
-		numErrsStripped int
-		kip320          = s.cl.supportsOffsetForLeaderEpoch()
+		updateWhy        multiUpdateWhy
+		debugWhyStripped multiUpdateWhy
+		numErrsStripped  int
+		kip320           = s.cl.supportsOffsetForLeaderEpoch()
 	)
 
+	strip := func(t string, p int32, err error) {
+		numErrsStripped++
+		if s.cl.cfg.logger.Level() < LogLevelDebug {
+			return
+		}
+		debugWhyStripped.add(t, p, err)
+	}
+
 	for _, rt := range resp.Topics {
 		topic := rt.Topic
 		// v13 only uses topic IDs, so we have to map the response
@@ -861,12 +870,12 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 			var keep bool
 			switch fp.Err {
 			default:
-				if kerr.IsRetriable(fp.Err) && !s.cl.cfg.keepFetchRetryableErrors {
+				if kerr.IsRetriable(fp.Err) && !s.cl.cfg.keepRetryableFetchErrors {
 					// UnknownLeaderEpoch: our meta is newer than the broker we fetched from
 					// OffsetNotAvailable: fetched from out of sync replica or a behind in-sync one (KIP-392 case 1 and case 2)
 					// UnknownTopicID: kafka has not synced the state on all brokers
 					// And other standard retryable errors.
-					numErrsStripped++
+					strip(topic, partition, fp.Err)
 				} else {
 					// - bad auth
 					// - unsupported compression
@@ -896,10 +905,10 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 				if fails := partOffset.from.unknownIDFails.Add(1); fails > 5 {
 					partOffset.from.unknownIDFails.Add(-1)
 					keep = true
-				} else if s.cl.cfg.keepFetchRetryableErrors {
+				} else if s.cl.cfg.keepRetryableFetchErrors {
 					keep = true
 				} else {
-					numErrsStripped++
+					strip(topic, partition, fp.Err)
 				}
 
 			case kerr.OffsetOutOfRange:
@@ -994,6 +1003,10 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 		}
 	}
 
+	if s.cl.cfg.logger.Level() >= LogLevelDebug && len(debugWhyStripped) > 0 {
+		s.cl.cfg.logger.Log(LogLevelDebug, "fetch stripped partitions", "why", debugWhyStripped.reason(""))
+	}
+
 	return f, reloadOffsets, preferreds, req.numOffsets == numErrsStripped, updateMeta, updateWhy.reason("fetch had inner topic errors")
 }
 
@@ -1835,7 +1848,7 @@ func (f *fetchRequest) MaxVersion() int16 {
 	if f.disableIDs {
 		return 12
 	}
-	return 13
+	return 15
 }
 func (f *fetchRequest) SetVersion(v int16) { f.version = v }
 func (f *fetchRequest) GetVersion() int16  { return f.version }
diff --git a/pkg/kgo/topics_and_partitions.go b/pkg/kgo/topics_and_partitions.go
index 9164e62c..ce1e8b83 100644
--- a/pkg/kgo/topics_and_partitions.go
+++ b/pkg/kgo/topics_and_partitions.go
@@ -108,6 +108,20 @@ func (m mtmps) onlyt(t string) bool {
 	return exists && len(ps) == 0
 }
 
+func (m mtmps) remove(t string, p int32) {
+	if m == nil {
+		return
+	}
+	mps, exists := m[t]
+	if !exists {
+		return
+	}
+	delete(mps, p)
+	if len(mps) == 0 {
+		delete(m, t)
+	}
+}
+
 ////////////
 // PAUSED // -- types for pausing topics and partitions
 ////////////
@@ -457,6 +471,8 @@ type topicPartitionsData struct {
 	isInternal         bool
 	partitions         []*topicPartition // partition num => partition
 	writablePartitions []*topicPartition // subset of above
+	topic              string
+	when               int64
 }
 
 // topicPartition contains all information from Kafka for a topic's partition,
diff --git a/pkg/kmsg/generated.go b/pkg/kmsg/generated.go
index 89face97..935279bd 100644
--- a/pkg/kmsg/generated.go
+++ b/pkg/kmsg/generated.go
@@ -3554,6 +3554,36 @@ func NewProduceResponse() ProduceResponse {
 	return v
 }
 
+type FetchRequestReplicaState struct {
+	// The replica ID of the follower, or -1 if this request is from a consumer.
+	//
+	// This field has a default of -1.
+	ID int32
+
+	// The epoch of this follower, or -1 if not available.
+	//
+	// This field has a default of -1.
+	Epoch int64
+
+	// UnknownTags are tags Kafka sent that we do not know the purpose of.
+	UnknownTags Tags // v12+
+}
+
+// Default sets any default fields. Calling this allows for future compatibility
+// if new fields are added to FetchRequestReplicaState.
+func (v *FetchRequestReplicaState) Default() {
+	v.ID = -1
+	v.Epoch = -1
+}
+
+// NewFetchRequestReplicaState returns a default FetchRequestReplicaState
+// This is a shortcut for creating a struct and calling Default yourself.
+func NewFetchRequestReplicaState() FetchRequestReplicaState {
+	var v FetchRequestReplicaState
+	v.Default()
+	return v
+}
+
 type FetchRequestTopicPartition struct {
 	// Partition is a partition in a topic to try to fetch records for.
 	Partition int32
@@ -3675,6 +3705,9 @@ func NewFetchRequestForgottenTopic() FetchRequestForgottenTopic {
 //
 // Starting in v13, topics must use UUIDs rather than their string name
 // identifiers.
+//
+// Version 15 adds the ReplicaState which includes new field ReplicaEpoch and
+// the ReplicaID, and deprecates the old ReplicaID (KIP-903).
 type FetchRequest struct {
 	// Version is the version of this message used with a Kafka broker.
 	Version int16
@@ -3688,7 +3721,12 @@ type FetchRequest struct {
 	// ReplicaID is the broker ID of performing the fetch request. Standard
 	// clients should use -1. To be a "debug" replica, use -2. The debug
 	// replica can be used to fetch messages from non-leaders.
-	ReplicaID int32
+	//
+	// This field has a default of -1.
+	ReplicaID int32 // v0-v14
+
+	// ReplicaState is a broker-only tag for v15+, see KIP-903 for more details.
+	ReplicaState FetchRequestReplicaState // tag 1
 
 	// MaxWaitMillis is how long to wait for MinBytes to be hit before a broker
 	// responds to a fetch request.
@@ -3748,7 +3786,7 @@ type FetchRequest struct {
 }
 
 func (*FetchRequest) Key() int16                 { return 1 }
-func (*FetchRequest) MaxVersion() int16          { return 13 }
+func (*FetchRequest) MaxVersion() int16          { return 15 }
 func (v *FetchRequest) SetVersion(version int16) { v.Version = version }
 func (v *FetchRequest) GetVersion() int16        { return v.Version }
 func (v *FetchRequest) IsFlexible() bool         { return v.Version >= 12 }
@@ -3772,7 +3810,7 @@ func (v *FetchRequest) AppendTo(dst []byte) []byte {
 	_ = version
 	isFlexible := version >= 12
 	_ = isFlexible
-	{
+	if version >= 0 && version <= 14 {
 		v := v.ReplicaID
 		dst = kbin.AppendInt32(dst, v)
 	}
@@ -3918,6 +3956,9 @@ func (v *FetchRequest) AppendTo(dst []byte) []byte {
 		if v.ClusterID != nil {
 			toEncode = append(toEncode, 0)
 		}
+		if !reflect.DeepEqual(v.ReplicaState, (func() FetchRequestReplicaState { var v FetchRequestReplicaState; v.Default(); return v })()) {
+			toEncode = append(toEncode, 1)
+		}
 		dst = kbin.AppendUvarint(dst, uint32(len(toEncode)+v.UnknownTags.Len()))
 		for _, tag := range toEncode {
 			switch tag {
@@ -3939,6 +3980,31 @@ func (v *FetchRequest) AppendTo(dst []byte) []byte {
 						goto fClusterID
 					}
 				}
+			case 1:
+				{
+					v := v.ReplicaState
+					dst = kbin.AppendUvarint(dst, 1)
+					sized := false
+					lenAt := len(dst)
+				fReplicaState:
+					{
+						v := v.ID
+						dst = kbin.AppendInt32(dst, v)
+					}
+					{
+						v := v.Epoch
+						dst = kbin.AppendInt64(dst, v)
+					}
+					if isFlexible {
+						dst = kbin.AppendUvarint(dst, 0+uint32(v.UnknownTags.Len()))
+						dst = v.UnknownTags.AppendEach(dst)
+					}
+					if !sized {
+						dst = kbin.AppendUvarint(dst[:lenAt], uint32(len(dst[lenAt:])))
+						sized = true
+						goto fReplicaState
+					}
+				}
 			}
 		}
 		dst = v.UnknownTags.AppendEach(dst)
@@ -3962,7 +4028,7 @@ func (v *FetchRequest) readFrom(src []byte, unsafe bool) error {
 	isFlexible := version >= 12
 	_ = isFlexible
 	s := v
-	{
+	if version >= 0 && version <= 14 {
 		v := b.Int32()
 		s.ReplicaID = v
 	}
@@ -4202,6 +4268,25 @@ func (v *FetchRequest) readFrom(src []byte, unsafe bool) error {
 				if err := b.Complete(); err != nil {
 					return err
 				}
+			case 1:
+				b := kbin.Reader{Src: b.Span(int(b.Uvarint()))}
+				v := &s.ReplicaState
+				v.Default()
+				s := v
+				{
+					v := b.Int32()
+					s.ID = v
+				}
+				{
+					v := b.Int64()
+					s.Epoch = v
+				}
+				if isFlexible {
+					s.UnknownTags = internalReadTags(&b)
+				}
+				if err := b.Complete(); err != nil {
+					return err
+				}
 			}
 		}
 	}
@@ -4220,6 +4305,13 @@ func NewPtrFetchRequest() *FetchRequest {
 // if new fields are added to FetchRequest.
 func (v *FetchRequest) Default() {
 	v.ClusterID = nil
+	v.ReplicaID = -1
+	{
+		v := &v.ReplicaState
+		_ = v
+		v.ID = -1
+		v.Epoch = -1
+	}
 	v.MaxBytes = 2147483647
 	v.SessionEpoch = -1
 }
@@ -4378,7 +4470,11 @@ type FetchResponseTopicPartition struct {
 	// OFFSET_OUT_OF_RANGE is returned if requesting an offset past the
 	// current end offset or before the beginning offset.
 	//
-	// UNKNOWN_TOPIC_ID is returned if using uuid's and the uuid is unknown.
+	// UNKNOWN_TOPIC_ID is returned if using uuid's and the uuid is unknown
+	// (v13+ / Kafka 3.1+).
+	//
+	// OFFSET_MOVED_TO_TIERED_STORAGE is returned if a follower is trying to
+	// fetch from an offset that is now in tiered storage.
 	ErrorCode int16
 
 	// HighWatermark is the current high watermark for this partition,
@@ -4544,7 +4640,7 @@ type FetchResponse struct {
 }
 
 func (*FetchResponse) Key() int16                         { return 1 }
-func (*FetchResponse) MaxVersion() int16                  { return 13 }
+func (*FetchResponse) MaxVersion() int16                  { return 15 }
 func (v *FetchResponse) SetVersion(version int16)         { v.Version = version }
 func (v *FetchResponse) GetVersion() int16                { return v.Version }
 func (v *FetchResponse) IsFlexible() bool                 { return v.Version >= 12 }
@@ -5070,6 +5166,11 @@ type ListOffsetsRequestTopicPartition struct {
 	// to determine topic "liveness" (when was the last produce?).
 	// Previously, this was not easy to determine. See KIP-734 for more
 	// detail.
+	//
+	// If you are talking to Kafka 3.4+ and using request version 8+ (for
+	// KIP-405), the new special timestamp -4 returns the local log start
+	// offset. In the context of tiered storage, the earliest local log start
+	// offset is the offset actually available on disk on the broker.
 	Timestamp int64
 
 	// MaxNumOffsets is the maximum number of offsets to report.
@@ -5130,6 +5231,9 @@ func NewListOffsetsRequestTopic() ListOffsetsRequestTopic {
 //
 // Version 7, introduced in Kafka 3.0, supports -3 as a timestamp to return
 // the timestamp and offset for the record with the largest timestamp.
+//
+// Version 8, introduced in Kafka 3.4, supports -4 as a timestamp to return
+// the local log start offset (in the context of tiered storage, see KIP-405).
 type ListOffsetsRequest struct {
 	// Version is the version of this message used with a Kafka broker.
 	Version int16
@@ -5157,7 +5261,7 @@ type ListOffsetsRequest struct {
 }
 
 func (*ListOffsetsRequest) Key() int16                 { return 2 }
-func (*ListOffsetsRequest) MaxVersion() int16          { return 7 }
+func (*ListOffsetsRequest) MaxVersion() int16          { return 8 }
 func (v *ListOffsetsRequest) SetVersion(version int16) { v.Version = version }
 func (v *ListOffsetsRequest) GetVersion() int16        { return v.Version }
 func (v *ListOffsetsRequest) IsFlexible() bool         { return v.Version >= 6 }
@@ -5522,7 +5626,7 @@ type ListOffsetsResponse struct {
 }
 
 func (*ListOffsetsResponse) Key() int16                         { return 2 }
-func (*ListOffsetsResponse) MaxVersion() int16                  { return 7 }
+func (*ListOffsetsResponse) MaxVersion() int16                  { return 8 }
 func (v *ListOffsetsResponse) SetVersion(version int16)         { v.Version = version }
 func (v *ListOffsetsResponse) GetVersion() int16                { return v.Version }
 func (v *ListOffsetsResponse) IsFlexible() bool                 { return v.Version >= 6 }
@@ -37567,6 +37671,33 @@ func NewDescribeQuorumResponse() DescribeQuorumResponse {
 	return v
 }
 
+type AlterPartitionRequestTopicPartitionNewEpochISR struct {
+	// The broker ID .
+	BrokerID int32
+
+	// The broker's epoch; -1 if the epoch check is not supported.
+	//
+	// This field has a default of -1.
+	BrokerEpoch int32
+
+	// UnknownTags are tags Kafka sent that we do not know the purpose of.
+	UnknownTags Tags
+}
+
+// Default sets any default fields. Calling this allows for future compatibility
+// if new fields are added to AlterPartitionRequestTopicPartitionNewEpochISR.
+func (v *AlterPartitionRequestTopicPartitionNewEpochISR) Default() {
+	v.BrokerEpoch = -1
+}
+
+// NewAlterPartitionRequestTopicPartitionNewEpochISR returns a default AlterPartitionRequestTopicPartitionNewEpochISR
+// This is a shortcut for creating a struct and calling Default yourself.
+func NewAlterPartitionRequestTopicPartitionNewEpochISR() AlterPartitionRequestTopicPartitionNewEpochISR {
+	var v AlterPartitionRequestTopicPartitionNewEpochISR
+	v.Default()
+	return v
+}
+
 type AlterPartitionRequestTopicPartition struct {
 	Partition int32
 
@@ -37574,7 +37705,9 @@ type AlterPartitionRequestTopicPartition struct {
 	LeaderEpoch int32
 
 	// The ISR for this partition.
-	NewISR []int32
+	NewISR []int32 // v0-v2
+
+	NewEpochISR []AlterPartitionRequestTopicPartitionNewEpochISR // v3+
 
 	// 1 if the partition is recovering from unclean leader election; 0 otherwise
 	LeaderRecoveryState int8 // v1+
@@ -37626,6 +37759,8 @@ func NewAlterPartitionRequestTopic() AlterPartitionRequestTopic {
 
 // AlterPartitionRequest, proposed in KIP-497 and introduced in Kafka 2.7.0,
 // is an admin request to modify ISR.
+//
+// Version 3 was added for KIP-903 and replaced NewISR.
 type AlterPartitionRequest struct {
 	// Version is the version of this message used with a Kafka broker.
 	Version int16
@@ -37645,7 +37780,7 @@ type AlterPartitionRequest struct {
 }
 
 func (*AlterPartitionRequest) Key() int16                 { return 56 }
-func (*AlterPartitionRequest) MaxVersion() int16          { return 2 }
+func (*AlterPartitionRequest) MaxVersion() int16          { return 3 }
 func (v *AlterPartitionRequest) SetVersion(version int16) { v.Version = version }
 func (v *AlterPartitionRequest) GetVersion() int16        { return v.Version }
 func (v *AlterPartitionRequest) IsFlexible() bool         { return v.Version >= 0 }
@@ -37716,7 +37851,7 @@ func (v *AlterPartitionRequest) AppendTo(dst []byte) []byte {
 						v := v.LeaderEpoch
 						dst = kbin.AppendInt32(dst, v)
 					}
-					{
+					if version >= 0 && version <= 2 {
 						v := v.NewISR
 						if isFlexible {
 							dst = kbin.AppendCompactArrayLen(dst, len(v))
@@ -37728,6 +37863,29 @@ func (v *AlterPartitionRequest) AppendTo(dst []byte) []byte {
 							dst = kbin.AppendInt32(dst, v)
 						}
 					}
+					if version >= 3 {
+						v := v.NewEpochISR
+						if isFlexible {
+							dst = kbin.AppendCompactArrayLen(dst, len(v))
+						} else {
+							dst = kbin.AppendArrayLen(dst, len(v))
+						}
+						for i := range v {
+							v := &v[i]
+							{
+								v := v.BrokerID
+								dst = kbin.AppendInt32(dst, v)
+							}
+							{
+								v := v.BrokerEpoch
+								dst = kbin.AppendInt32(dst, v)
+							}
+							if isFlexible {
+								dst = kbin.AppendUvarint(dst, 0+uint32(v.UnknownTags.Len()))
+								dst = v.UnknownTags.AppendEach(dst)
+							}
+						}
+					}
 					if version >= 1 {
 						v := v.LeaderRecoveryState
 						dst = kbin.AppendInt8(dst, v)
@@ -37848,7 +38006,7 @@ func (v *AlterPartitionRequest) readFrom(src []byte, unsafe bool) error {
 						v := b.Int32()
 						s.LeaderEpoch = v
 					}
-					{
+					if version >= 0 && version <= 2 {
 						v := s.NewISR
 						a := v
 						var l int32
@@ -37871,6 +38029,41 @@ func (v *AlterPartitionRequest) readFrom(src []byte, unsafe bool) error {
 						v = a
 						s.NewISR = v
 					}
+					if version >= 3 {
+						v := s.NewEpochISR
+						a := v
+						var l int32
+						if isFlexible {
+							l = b.CompactArrayLen()
+						} else {
+							l = b.ArrayLen()
+						}
+						if !b.Ok() {
+							return b.Complete()
+						}
+						a = a[:0]
+						if l > 0 {
+							a = append(a, make([]AlterPartitionRequestTopicPartitionNewEpochISR, l)...)
+						}
+						for i := int32(0); i < l; i++ {
+							v := &a[i]
+							v.Default()
+							s := v
+							{
+								v := b.Int32()
+								s.BrokerID = v
+							}
+							{
+								v := b.Int32()
+								s.BrokerEpoch = v
+							}
+							if isFlexible {
+								s.UnknownTags = internalReadTags(&b)
+							}
+						}
+						v = a
+						s.NewEpochISR = v
+					}
 					if version >= 1 {
 						v := b.Int8()
 						s.LeaderRecoveryState = v
@@ -38000,7 +38193,7 @@ type AlterPartitionResponse struct {
 }
 
 func (*AlterPartitionResponse) Key() int16                         { return 56 }
-func (*AlterPartitionResponse) MaxVersion() int16                  { return 2 }
+func (*AlterPartitionResponse) MaxVersion() int16                  { return 3 }
 func (v *AlterPartitionResponse) SetVersion(version int16)         { v.Version = version }
 func (v *AlterPartitionResponse) GetVersion() int16                { return v.Version }
 func (v *AlterPartitionResponse) IsFlexible() bool                 { return v.Version >= 0 }
diff --git a/pkg/kversion/kversion.go b/pkg/kversion/kversion.go
index 6582db44..a70d1cbb 100644
--- a/pkg/kversion/kversion.go
+++ b/pkg/kversion/kversion.go
@@ -179,7 +179,78 @@ type guessCfg struct {
 // Options can be specified to change how version guessing is performed, for
 // example, certain keys can be skipped, or the guessing can try evaluating the
 // versions as Raft broker based versions.
+//
+// Internally, this function tries guessing the version against both KRaft and
+// Kafka APIs. The more exact match is returned.
 func (vs *Versions) VersionGuess(opts ...VersionGuessOpt) string {
+	standard := vs.versionGuess(opts...)
+	raftBroker := vs.versionGuess(append(opts, TryRaftBroker())...)
+	raftController := vs.versionGuess(append(opts, TryRaftController())...)
+
+	// If any of these are exact, return the exact guess.
+	for _, g := range []guess{
+		standard,
+		raftBroker,
+		raftController,
+	} {
+		if g.how == guessExact {
+			return g.String()
+		}
+	}
+
+	// If any are atLeast, that means it is newer than we can guess and we
+	// return the highest version.
+	for _, g := range []guess{
+		standard,
+		raftBroker,
+		raftController,
+	} {
+		if g.how == guessAtLeast {
+			return g.String()
+		}
+	}
+
+	// This is a custom version. We could do some advanced logic to try to
+	// return highest of all three guesses, but that may be inaccurate:
+	// KRaft may detect a higher guess because not all requests exist in
+	// KRaft. Instead, we just return our standard guess.
+	return standard.String()
+}
+
+type guess struct {
+	v1  string
+	v2  string // for between
+	how int8
+}
+
+const (
+	guessExact = iota
+	guessAtLeast
+	guessCustomUnknown
+	guessCustomAtLeast
+	guessBetween
+	guessNotEven
+)
+
+func (g guess) String() string {
+	switch g.how {
+	case guessExact:
+		return g.v1
+	case guessAtLeast:
+		return "at least " + g.v1
+	case guessCustomUnknown:
+		return "unknown custom version"
+	case guessCustomAtLeast:
+		return "unknown custom version at least " + g.v1
+	case guessBetween:
+		return "between " + g.v1 + " and " + g.v2
+	case guessNotEven:
+		return "not even " + g.v1
+	}
+	return g.v1
+}
+
+func (vs *Versions) versionGuess(opts ...VersionGuessOpt) guess {
 	cfg := guessCfg{
 		listener: zkBroker,
 		skipKeys: []int16{4, 5, 6, 7, 27},
@@ -224,6 +295,7 @@ func (vs *Versions) VersionGuess(opts ...VersionGuessOpt) string {
 		{max320, "v3.2"},
 		{max330, "v3.3"},
 		{max340, "v3.4"},
+		{max350, "v3.5"},
 	} {
 		for k, v := range comparison.cmp.filter(cfg.listener) {
 			if v == -1 {
@@ -286,30 +358,30 @@ func (vs *Versions) VersionGuess(opts ...VersionGuessOpt) string {
 		case under && over:
 			// Regardless of equal being true or not, this is a custom version.
 			if last != "" {
-				return "unknown custom version at least " + last
+				return guess{v1: last, how: guessCustomAtLeast}
 			}
-			return "unknown custom version"
+			return guess{v1: last, how: guessCustomUnknown}
 
 		case under:
 			// Regardless of equal being true or not, we have not yet hit
 			// this version.
 			if last != "" {
-				return "between " + last + " and " + current
+				return guess{v1: last, v2: current, how: guessBetween}
 			}
-			return "not even " + current
+			return guess{v1: current, how: guessNotEven}
 
 		case over:
 			// Regardless of equal being true or not, we try again.
 			last = current
 
 		case equal:
-			return current
+			return guess{v1: current, how: guessExact}
 		}
 		// At least one of under, equal, or over must be true, so there
 		// is no default case.
 	}
 
-	return "at least " + last
+	return guess{v1: last, how: guessAtLeast}
 }
 
 // String returns a string representation of the versions; the format may
@@ -363,6 +435,7 @@ func V3_1_0() *Versions  { return zkBrokerOf(max310) }
 func V3_2_0() *Versions  { return zkBrokerOf(max320) }
 func V3_3_0() *Versions  { return zkBrokerOf(max330) }
 func V3_4_0() *Versions  { return zkBrokerOf(max340) }
+func V3_5_0() *Versions  { return zkBrokerOf(max350) }
 
 func zkBrokerOf(lks listenerKeys) *Versions {
 	return &Versions{lks.filter(zkBroker)}
@@ -506,7 +579,7 @@ var max0110 = nextMax(max0102, func(v listenerKeys) listenerKeys {
 		k(zkBroker, rBroker), // 23 offset for leader epoch KAFKA-1211 0baea2ac13 KIP-101
 
 		k(zkBroker, rBroker), // 24 add partitions to txn KAFKA-4990 865d82af2c KIP-98 (raft 3.0 6e857c531f14d07d5b05f174e6063a124c917324)
-		k(zkBroker),          // 25 add offsets to txn (same, same raft)
+		k(zkBroker, rBroker), // 25 add offsets to txn (same, same raft)
 		k(zkBroker, rBroker), // 26 end txn (same, same raft)
 		k(zkBroker, rBroker), // 27 write txn markers (same)
 		k(zkBroker, rBroker), // 28 txn offset commit (same, same raft)
@@ -767,8 +840,8 @@ var max270 = nextMax(max260, func(v listenerKeys) listenerKeys {
 	v[26].inc() // 2 end txn
 
 	v = append(v,
-		k(zkBroker), // 50 describe user scram creds, KAFKA-10259 e8524ccd8fca0caac79b844d87e98e9c055f76fb KIP-554
-		k(zkBroker), // 51 alter user scram creds, same
+		k(zkBroker, rBroker, rController), // 50 describe user scram creds, KAFKA-10259 e8524ccd8fca0caac79b844d87e98e9c055f76fb KIP-554; 38c409cf33c kraft
+		k(zkBroker, rBroker, rController), // 51 alter user scram creds, same
 	)
 
 	// KAFKA-10435 634c9175054cc69d10b6da22ea1e95edff6a4747 KIP-595
@@ -960,9 +1033,19 @@ var max340 = nextMax(max330, func(v listenerKeys) listenerKeys {
 	return v
 })
 
+var max350 = nextMax(max340, func(v listenerKeys) listenerKeys {
+	// KAFKA-13369 7146ac57ba9ddd035dac992b9f188a8e7677c08d KIP-405
+	v[1].inc() // 14 fetch
+	v[2].inc() // 8 list offsets
+
+	v[1].inc()  // 15 fetch // KAFKA-14617 79b5f7f1ce2 KIP-903
+	v[56].inc() // 3 alter partition // KAFKA-14617 8c88cdb7186b1d594f991eb324356dcfcabdf18a KIP-903
+	return v
+})
+
 var (
-	maxStable = max340
+	maxStable = max350
 	maxTip    = nextMax(maxStable, func(v listenerKeys) listenerKeys {
-		return maxStable
+		return v
 	})
 )
diff --git a/pkg/sasl/oauth/oauth.go b/pkg/sasl/oauth/oauth.go
index 31f6a8dc..6527c076 100644
--- a/pkg/sasl/oauth/oauth.go
+++ b/pkg/sasl/oauth/oauth.go
@@ -52,6 +52,9 @@ func (fn oauth) Authenticate(ctx context.Context, _ string) (sasl.Session, []byt
 	if err != nil {
 		return nil, nil, err
 	}
+	if auth.Token == "" {
+		return nil, nil, errors.New("OAUTHBEARER token must be non-empty")
+	}
 
 	// We sort extensions for consistency, but it is not required.
 	type kv struct {
diff --git a/pkg/sasl/plain/plain.go b/pkg/sasl/plain/plain.go
index fbcc2ed0..97a9369d 100644
--- a/pkg/sasl/plain/plain.go
+++ b/pkg/sasl/plain/plain.go
@@ -3,6 +3,7 @@ package plain
 
 import (
 	"context"
+	"errors"
 
 	"github.com/twmb/franz-go/pkg/sasl"
 )
@@ -46,6 +47,9 @@ func (fn plain) Authenticate(ctx context.Context, _ string) (sasl.Session, []byt
 	if err != nil {
 		return nil, nil, err
 	}
+	if auth.User == "" || auth.Pass == "" {
+		return nil, nil, errors.New("PLAIN user and pass must be non-empty")
+	}
 	return session{}, []byte(auth.Zid + "\x00" + auth.User + "\x00" + auth.Pass), nil
 }
 
diff --git a/pkg/sasl/sasl.go b/pkg/sasl/sasl.go
index 4065e7e5..dd85a02a 100644
--- a/pkg/sasl/sasl.go
+++ b/pkg/sasl/sasl.go
@@ -1,4 +1,4 @@
-// Package sasl specifies interfaces that any sasl authentication must provide
+// Package sasl specifies interfaces that any SASL authentication must provide
 // to interop with Kafka SASL.
 package sasl
 
@@ -32,7 +32,7 @@ type Mechanism interface {
 	Authenticate(ctx context.Context, host string) (Session, []byte, error)
 }
 
-// ClosingMechanism is an optional interface for sasl mechanism's. Implementing
+// ClosingMechanism is an optional interface for SASL mechanisms. Implementing
 // this interface signals that the mechanism should be closed if it will never
 // be used again.
 type ClosingMechanism interface {
diff --git a/pkg/sasl/scram/scram.go b/pkg/sasl/scram/scram.go
index 05f785fb..6272da66 100644
--- a/pkg/sasl/scram/scram.go
+++ b/pkg/sasl/scram/scram.go
@@ -105,6 +105,9 @@ func (s scram) Authenticate(ctx context.Context, _ string) (sasl.Session, []byte
 	if err != nil {
 		return nil, nil, err
 	}
+	if auth.User == "" || auth.Pass == "" {
+		return nil, nil, errors.New(s.name + " user and pass must be non-empty")
+	}
 	if len(auth.Nonce) == 0 {
 		buf := make([]byte, 20)
 		if _, err = rand.Read(buf); err != nil {
diff --git a/pkg/sr/serde.go b/pkg/sr/serde.go
index 3f95af01..8de7b5eb 100644
--- a/pkg/sr/serde.go
+++ b/pkg/sr/serde.go
@@ -9,11 +9,6 @@ import (
 	"sync/atomic"
 )
 
-// The wire format for encoded types is 0, then big endian uint32 of the ID,
-// then the encoded message.
-//
-// https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format
-
 var (
 	// ErrNotRegistered is returned from Serde when attempting to encode a
 	// value or decode an ID that has not been registered, or when using
@@ -22,7 +17,7 @@ var (
 
 	// ErrBadHeader is returned from Decode when the input slice is shorter
 	// than five bytes, or if the first byte is not the magic 0 byte.
-	ErrBadHeader = errors.New("5 byte header for value is missing or does no have 0 magic byte")
+	ErrBadHeader = errors.New("5 byte header for value is missing or does not have 0 magic byte")
 )
 
 type (
@@ -80,8 +75,9 @@ type tserde struct {
 	gen          func() any
 	typeof       reflect.Type
 
-	index    []int          // for encoding, an optional index we use
-	subindex map[int]tserde // for decoding, we look up sub-indices in the payload
+	index         []int          // for encoding, an optional index we use
+	subindex      map[int]tserde // for decoding, we look up sub-indices in the payload
+	subindexDepth int            // for decoding, we need to know the maximum depth of the subindex map
 }
 
 // Serde encodes and decodes values according to the schema registry wire
@@ -101,6 +97,7 @@ type Serde struct {
 	mu    sync.Mutex
 
 	defaults []SerdeOpt
+	h        SerdeHeader
 }
 
 var (
@@ -134,6 +131,26 @@ func (s *Serde) SetDefaults(opts ...SerdeOpt) {
 	s.defaults = opts
 }
 
+// DecodeID decodes an ID from b, returning the ID and the remaining bytes,
+// or an error.
+func (s *Serde) DecodeID(b []byte) (id int, out []byte, err error) {
+	return s.header().DecodeID(b)
+}
+
+// DecodeIndex decodes at most maxLength of a schema index from in, returning
+// the index and remaining bytes, or an error. It expects b to be the output of
+// DecodeID (schema ID should already be stripped away).
+func (s *Serde) DecodeIndex(in []byte, maxLength int) (index []int, out []byte, err error) {
+	return s.header().DecodeIndex(in, maxLength)
+}
+
+func (s *Serde) header() SerdeHeader {
+	if s.h == nil {
+		return defaultSerdeHeader
+	}
+	return s.h
+}
+
 // Register registers a schema ID and the value it corresponds to, as well as
 // the encoding or decoding functions. You need to register functions depending
 // on whether you are only encoding, only decoding, or both.
@@ -152,27 +169,42 @@ func (s *Serde) Register(id int, v any, opts ...SerdeOpt) {
 	defer s.mu.Unlock()
 
 	// Type mapping is easy: we just map the type to the final tserde.
-	// We defer the store because we modify the tserde below, and we
-	// may delete a type key.
 	dupTypes := make(map[reflect.Type]tserde)
 	for k, v := range s.loadTypes() {
 		dupTypes[k] = v
 	}
+
+	// For IDs, we deeply clone any path that is changing.
+	dupIDs := tserdeMapClone(s.loadIDs(), id, t.index)
+
+	// We defer the store because we modify the tserde below, and we
+	// may delete a type key.
 	defer func() {
 		dupTypes[typeof] = t
 		s.types.Store(dupTypes)
+		s.ids.Store(dupIDs)
 	}()
 
-	// For IDs, we deeply clone any path that is changing.
-	m := tserdeMapClone(s.loadIDs(), id, t.index)
-	s.ids.Store(m)
-
 	// Now we have a full path down index initialized (or, the top
 	// level map if there is no index). We iterate down the index
 	// tree to find the end node we are initializing.
 	k := id
+	m := dupIDs
 	at := m[k]
+	depth := len(t.index)
+	max := func(i, j int) int {
+		if i > j {
+			return i
+		}
+		return j
+	}
 	for _, idx := range t.index {
+		// SAFETY: tserdeMapClone deeply clones all maps through the index, so
+		// our modified value is being saved to a new map that is not being read.
+		at.subindexDepth = max(at.subindexDepth, depth)
+		m[k] = at
+		depth--
+
 		m = at.subindex
 		k = idx
 		at = m[k]
@@ -186,15 +218,16 @@ func (s *Serde) Register(id int, v any, opts ...SerdeOpt) {
 
 	// Now, we initialize the end node.
 	t = tserde{
-		id:           uint32(id),
-		exists:       true,
-		encode:       t.encode,
-		appendEncode: t.appendEncode,
-		decode:       t.decode,
-		gen:          t.gen,
-		typeof:       typeof,
-		index:        t.index,
-		subindex:     at.subindex,
+		id:            uint32(id),
+		exists:        true,
+		encode:        t.encode,
+		appendEncode:  t.appendEncode,
+		decode:        t.decode,
+		gen:           t.gen,
+		typeof:        typeof,
+		index:         t.index,
+		subindex:      at.subindex,
+		subindexDepth: at.subindexDepth,
 	}
 	m[k] = t
 }
@@ -219,8 +252,8 @@ func tserdeMapClone(m map[int]tserde, at int, index []int) map[int]tserde {
 	return dup
 }
 
-// Encode encodes a value according to the schema registry wire format and
-// returns it. If EncodeFn was not used, this returns ErrNotRegistered.
+// Encode encodes a value and prepends the header according to the configured
+// SerdeHeader. If EncodeFn was not used, this returns ErrNotRegistered.
 func (s *Serde) Encode(v any) ([]byte, error) {
 	return s.AppendEncode(nil, v)
 }
@@ -234,23 +267,9 @@ func (s *Serde) AppendEncode(b []byte, v any) ([]byte, error) {
 		return b, ErrNotRegistered
 	}
 
-	b = append(b,
-		0,
-		byte(t.id>>24),
-		byte(t.id>>16),
-		byte(t.id>>8),
-		byte(t.id>>0),
-	)
-
-	if len(t.index) > 0 {
-		if len(t.index) == 1 && t.index[0] == 0 {
-			b = append(b, 0) // first-index shortcut (one type in the protobuf)
-		} else {
-			b = binary.AppendVarint(b, int64(len(t.index)))
-			for _, idx := range t.index {
-				b = binary.AppendVarint(b, int64(idx))
-			}
-		}
+	b, err := s.header().AppendEncode(b, int(t.id), t.index)
+	if err != nil {
+		return nil, err
 	}
 
 	if t.appendEncode != nil {
@@ -316,37 +335,122 @@ func (s *Serde) DecodeNew(b []byte) (any, error) {
 }
 
 func (s *Serde) decodeFind(b []byte) ([]byte, tserde, error) {
-	if len(b) < 5 || b[0] != 0 {
-		return nil, tserde{}, ErrBadHeader
+	id, b, err := s.DecodeID(b)
+	if err != nil {
+		return nil, tserde{}, err
 	}
-	id := binary.BigEndian.Uint32(b[1:5])
-	b = b[5:]
 
-	t := s.loadIDs()[int(id)]
+	t := s.loadIDs()[id]
 	if len(t.subindex) > 0 {
-		r := bReader{b}
-		br := io.ByteReader(&r)
-		l, err := binary.ReadVarint(br)
-		if l == 0 { // length 0 is a shortcut for length 1, index 0
-			t = t.subindex[0]
-		}
-		for err == nil && t.subindex != nil && l > 0 {
-			var idx int64
-			idx, err = binary.ReadVarint(br)
-			t = t.subindex[int(idx)]
-			l--
-		}
+		var index []int
+		index, b, err = s.DecodeIndex(b, t.subindexDepth)
 		if err != nil {
-			return nil, t, err
+			return nil, tserde{}, err
+		}
+		for _, idx := range index {
+			if t.subindex == nil {
+				return nil, tserde{}, ErrNotRegistered
+			}
+			t = t.subindex[idx]
 		}
-		b = r.b
 	}
 	if !t.exists {
-		return nil, t, ErrNotRegistered
+		return nil, tserde{}, ErrNotRegistered
 	}
 	return b, t, nil
 }
 
+// SerdeHeader encodes and decodes a message header.
+type SerdeHeader interface {
+	// AppendEncode encodes a schema ID and optional index to b, returning the
+	// updated slice or an error.
+	AppendEncode(b []byte, id int, index []int) ([]byte, error)
+	// DecodeID decodes an ID from in, returning the ID and the remaining bytes,
+	// or an error.
+	DecodeID(in []byte) (id int, out []byte, err error)
+	// DecodeIndex decodes at most maxLength of a schema index from in,
+	// returning the index and remaining bytes, or an error.
+	DecodeIndex(in []byte, maxLength int) (index []int, out []byte, err error)
+}
+
+var defaultSerdeHeader = new(confluentHeader)
+
+// confluentHeader is a SerdeHeader that produces the Confluent wire format. It
+// starts with 0, then big endian uint32 of the ID, then index (only protobuf),
+// then the encoded message.
+//
+// https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format
+type confluentHeader struct{}
+
+// AppendEncode appends an encoded header to b according to the Confluent wire
+// format and returns it. Error is always nil.
+func (*confluentHeader) AppendEncode(b []byte, id int, index []int) ([]byte, error) {
+	b = append(
+		b,
+		0,
+		byte(id>>24),
+		byte(id>>16),
+		byte(id>>8),
+		byte(id>>0),
+	)
+
+	if len(index) > 0 {
+		if len(index) == 1 && index[0] == 0 {
+			b = append(b, 0) // first-index shortcut (one type in the protobuf)
+		} else {
+			b = binary.AppendVarint(b, int64(len(index)))
+			for _, idx := range index {
+				b = binary.AppendVarint(b, int64(idx))
+			}
+		}
+	}
+
+	return b, nil
+}
+
+// DecodeID strips and decodes the schema ID from b. It returns the ID alongside
+// the unread bytes. If the header does not contain the magic byte or b contains
+// less than 5 bytes it returns ErrBadHeader.
+func (*confluentHeader) DecodeID(b []byte) (int, []byte, error) {
+	if len(b) < 5 || b[0] != 0 {
+		return 0, nil, ErrBadHeader
+	}
+	id := binary.BigEndian.Uint32(b[1:5])
+	return int(id), b[5:], nil
+}
+
+// DecodeIndex strips and decodes indices from b. It returns the index slice
+// alongside the unread bytes. It expects b to be the output of DecodeID (schema
+// ID should already be stripped away). If maxLength is greater than 0 and the
+// encoded data contains more indices than maxLength the function returns
+// ErrNotRegistered.
+func (*confluentHeader) DecodeIndex(b []byte, maxLength int) ([]int, []byte, error) {
+	r := bReader{b}
+	br := io.ByteReader(&r)
+	l, err := binary.ReadVarint(br)
+	if err != nil {
+		return nil, nil, err
+	}
+	if l == 0 { // length 0 is a shortcut for length 1, index 0
+		return []int{0}, r.b, nil
+	}
+	if l < 0 { // index length can't be negative
+		return nil, nil, ErrBadHeader
+	}
+	if maxLength > 0 && int(l) > maxLength { // index count is greater than expected
+		return nil, nil, ErrNotRegistered
+	}
+	index := make([]int, l)
+	for i := range index {
+		idx, err := binary.ReadVarint(br)
+		if err != nil {
+			return nil, nil, err
+		}
+		index[i] = int(idx)
+	}
+	return index, r.b, nil
+}
+
 type bReader struct{ b []byte }
 
 func (b *bReader) ReadByte() (byte, error) {
diff --git a/plugin/README.md b/plugin/README.md
index 1e835feb..fc0c9581 100644
--- a/plugin/README.md
+++ b/plugin/README.md
@@ -13,6 +13,7 @@ Each of these plugins has a corresponding example in [examples](../examples/hook
 <a href="./">plugin</a> — you are here
 ├── <a href="./kgmetrics">kgmetrics</a> — plug-in go-metrics to use with `kgo.WithHooks`
 ├── <a href="./kprom">kprom</a> — plug-in prometheus metrics to use with `kgo.WithHooks`
+├── <a href="./klogrus">klogrus</a> — plug-in sirupsen/logrus to use with `kgo.WithLogger`
 ├── <a href="./kzap">kzap</a> — plug-in uber-go/zap to use with `kgo.WithLogger`
 └── <a href="./kzerolog">kzerolog</a> — plug-in rs/zerolog to use with `kgo.WithLogger`
 </pre>
diff --git a/plugin/klogrus/README.md b/plugin/klogrus/README.md
new file mode 100644
index 00000000..47b33bfb
--- /dev/null
+++ b/plugin/klogrus/README.md
@@ -0,0 +1,13 @@
+klogrus
+===
+
+klogrus is a plug-in package to use [logrus](https://github.com/sirupsen/logrus) as a [`kgo.Logger`](https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo#Logger)
+
+To use,
+
+```go
+cl, err := kgo.NewClient(
+        kgo.WithLogger(klogrus.New(logrusLogger)),
+        // ...other opts
+)
+```
\ No newline at end of file
diff --git a/plugin/klogrus/go.mod b/plugin/klogrus/go.mod
new file mode 100644
index 00000000..fb5cf51f
--- /dev/null
+++ b/plugin/klogrus/go.mod
@@ -0,0 +1,16 @@
+module github.com/twmb/franz-go/plugin/klogrus
+
+go 1.18
+
+require (
+	github.com/sirupsen/logrus v1.9.3
+	github.com/twmb/franz-go v1.13.5
+)
+
+require (
+	github.com/klauspost/compress v1.16.3 // indirect
+	github.com/pierrec/lz4/v4 v4.1.17 // indirect
+	github.com/stretchr/testify v1.8.0 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.4.0 // indirect
+	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
+)
diff --git a/plugin/klogrus/go.sum b/plugin/klogrus/go.sum
new file mode 100644
index 00000000..fa9da63b
--- /dev/null
+++ b/plugin/klogrus/go.sum
@@ -0,0 +1,27 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/klauspost/compress v1.16.3 h1:XuJt9zzcnaz6a16/OU53ZjWp/v7/42WcR5t2a0PcNQY=
+github.com/klauspost/compress v1.16.3/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
+github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/twmb/franz-go v1.13.5 h1:7Hk47eZ7XRb4yWXQZk1GZU4BthkrKuZUfKOuP9Sgp24=
+github.com/twmb/franz-go v1.13.5/go.mod h1:jm/FtYxmhxDTN0gNSb26XaJY0irdSVcsckLiR5tQNMk=
+github.com/twmb/franz-go/pkg/kmsg v1.4.0 h1:tbp9hxU6m8qZhQTlpGiaIJOm4BXix5lsuEZ7K00dF0s=
+github.com/twmb/franz-go/pkg/kmsg v1.4.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/plugin/klogrus/klogrus.go b/plugin/klogrus/klogrus.go
new file mode 100644
index 00000000..70ec6364
--- /dev/null
+++ b/plugin/klogrus/klogrus.go
@@ -0,0 +1,65 @@
+package klogrus
+
+import (
+	"fmt"
+
+	"github.com/sirupsen/logrus"
+	"github.com/twmb/franz-go/pkg/kgo"
+)
+
+// Logger provides the kgo.Logger interface for usage in kgo.WithLogger when
+// initializing a client.
+type Logger struct {
+	lr *logrus.Logger
+}
+
+// New returns a new Logger.
+func New(lr *logrus.Logger) *Logger {
+	return &Logger{lr}
+}
+
+// Level is for the kgo.Logger interface.
+func (l *Logger) Level() kgo.LogLevel {
+	return logrusToKgoLevel(l.lr.GetLevel())
+}
+
+// Log is for the kgo.Logger interface.
+func (l *Logger) Log(level kgo.LogLevel, msg string, keyvals ...any) {
+	logrusLevel, levelMatched := kgoToLogrusLevel(level)
+	if levelMatched {
+		fields := make(logrus.Fields, len(keyvals)/2)
+		for i := 0; i < len(keyvals); i += 2 {
+			fields[fmt.Sprint(keyvals[i])] = keyvals[i+1]
+		}
+		l.lr.WithFields(fields).Log(logrusLevel, msg)
+	}
+}
+
+func kgoToLogrusLevel(level kgo.LogLevel) (logrus.Level, bool) {
+	switch level {
+	case kgo.LogLevelError:
+		return logrus.ErrorLevel, true
+	case kgo.LogLevelWarn:
+		return logrus.WarnLevel, true
+	case kgo.LogLevelInfo:
+		return logrus.InfoLevel, true
+	case kgo.LogLevelDebug:
+		return logrus.DebugLevel, true
+	}
+	return logrus.TraceLevel, false
+}
+
+func logrusToKgoLevel(level logrus.Level) kgo.LogLevel {
+	switch level {
+	case logrus.PanicLevel, logrus.FatalLevel, logrus.ErrorLevel:
+		return kgo.LogLevelError
+	case logrus.WarnLevel:
+		return kgo.LogLevelWarn
+	case logrus.InfoLevel:
+		return kgo.LogLevelInfo
+	case logrus.DebugLevel, logrus.TraceLevel:
+		return kgo.LogLevelDebug
+	default:
+		return kgo.LogLevelNone
+	}
+}


Test patch:
diff --git a/pkg/kgo/consumer_direct_test.go b/pkg/kgo/consumer_direct_test.go
index 72f78625..8eadf5da 100644
--- a/pkg/kgo/consumer_direct_test.go
+++ b/pkg/kgo/consumer_direct_test.go
@@ -3,6 +3,7 @@ package kgo
 import (
 	"context"
 	"fmt"
+	"sort"
 	"testing"
 	"time"
 )
@@ -158,7 +159,7 @@ func TestIssue434(t *testing.T) {
 		ConsumeTopics(fmt.Sprintf("(%s|%s)", t1, t2)),
 		ConsumeRegex(),
 		FetchMaxWait(100*time.Millisecond),
-		keepFetchRetryableErrors(),
+		KeepRetryableFetchErrors(),
 	)
 	defer cl.Close()
 
@@ -181,8 +182,8 @@ func TestIssue434(t *testing.T) {
 	start := time.Now()
 	var missingTopic int
 	for missingTopic < 2 {
-		if time.Since(start) > 2*time.Second {
-			t.Fatal("still seeing topic after 2s")
+		if time.Since(start) > 30*time.Second {
+			t.Fatal("still seeing topic after 30s")
 		}
 
 		ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
@@ -199,3 +200,121 @@ func TestIssue434(t *testing.T) {
 		}
 	}
 }
+
+func TestAddRemovePartitions(t *testing.T) {
+	t.Parallel()
+
+	t1, cleanup := tmpTopicPartitions(t, 2)
+	defer cleanup()
+
+	cl, _ := NewClient(
+		getSeedBrokers(),
+		UnknownTopicRetries(-1),
+		RecordPartitioner(ManualPartitioner()),
+		FetchMaxWait(100*time.Millisecond),
+	)
+	defer cl.Close()
+
+	if err := cl.ProduceSync(context.Background(),
+		&Record{Topic: t1, Partition: 0, Value: []byte("v1")},
+		&Record{Topic: t1, Partition: 1, Value: []byte("v2")},
+		&Record{Topic: t1, Partition: 1, Value: []byte("v3")},
+	).FirstErr(); err != nil {
+		t.Fatal(err)
+	}
+
+	cl.AddConsumePartitions(map[string]map[int32]Offset{
+		t1: {0: NewOffset().At(0)},
+	})
+
+	recs := cl.PollFetches(context.Background()).Records()
+	if len(recs) != 1 || string(recs[0].Value) != "v1" {
+		t.Fatalf("expected to see v1, got %v", recs)
+	}
+
+	cl.RemoveConsumePartitions(map[string][]int32{
+		t1:   {0, 1, 2},
+		"t2": {0, 1, 2},
+	})
+
+	cl.AddConsumePartitions(map[string]map[int32]Offset{
+		t1: {
+			0: NewOffset().At(0),
+			1: NewOffset().At(1),
+		},
+	})
+
+	recs = recs[:0]
+	for len(recs) < 2 {
+		recs = append(recs, cl.PollFetches(context.Background()).Records()...)
+	}
+	if len(recs) > 2 {
+		t.Fatalf("expected to see 2 records, got %v", recs)
+	}
+
+	sort.Slice(recs, func(i, j int) bool {
+		return recs[i].Partition < recs[j].Partition
+	})
+
+	if string(recs[0].Value) != "v1" || string(recs[1].Value) != "v3" {
+		t.Fatalf("expected to see v1 and v2, got %v", recs)
+	}
+}
+
+func TestPauseIssue489(t *testing.T) {
+	t.Parallel()
+
+	t1, cleanup := tmpTopicPartitions(t, 2)
+	defer cleanup()
+
+	cl, _ := NewClient(
+		getSeedBrokers(),
+		UnknownTopicRetries(-1),
+		DefaultProduceTopic(t1),
+		RecordPartitioner(ManualPartitioner()),
+		ConsumeTopics(t1),
+		FetchMaxWait(100*time.Millisecond),
+	)
+	defer cl.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		var exit bool
+		var zeroOne uint8
+		for !exit {
+			r := StringRecord("v")
+			r.Partition = int32(zeroOne % 2)
+			zeroOne++
+			cl.Produce(ctx, r, func(r *Record, err error) {
+				if err == context.Canceled {
+					exit = true
+				}
+			})
+		}
+	}()
+	defer cancel()
+
+	for i := 0; i < 10; i++ {
+		var sawZero, sawOne bool
+		for !sawZero || !sawOne {
+			fs := cl.PollFetches(ctx)
+			fs.EachRecord(func(r *Record) {
+				sawZero = sawZero || r.Partition == 0
+				sawOne = sawOne || r.Partition == 1
+			})
+		}
+		cl.PauseFetchPartitions(map[string][]int32{t1: {0}})
+		sawZero, sawOne = false, false
+		for i := 0; i < 5; i++ {
+			fs := cl.PollFetches(ctx)
+			fs.EachRecord(func(r *Record) {
+				sawZero = sawZero || r.Partition == 0
+				sawOne = sawOne || r.Partition == 1
+			})
+		}
+		if sawZero {
+			t.Error("saw partition zero even though it was paused")
+		}
+		cl.ResumeFetchPartitions(map[string][]int32{t1: {0}})
+	}
+}
diff --git a/pkg/sr/serde_test.go b/pkg/sr/serde_test.go
index 8667a0d2..45287c30 100644
--- a/pkg/sr/serde_test.go
+++ b/pkg/sr/serde_test.go
@@ -135,7 +135,86 @@ func TestSerde(t *testing.T) {
 	if _, err := serde.DecodeNew([]byte{0, 0, 0, 0, 3}); err != io.EOF {
 		t.Errorf("got %v != exp io.EOF", err)
 	}
+	if _, err := serde.DecodeNew([]byte{0, 0, 0, 0, 3, 8}); err != ErrNotRegistered {
+		t.Errorf("got %v != exp ErrNotRegistered", err)
+	}
 	if _, err := serde.DecodeNew([]byte{0, 0, 0, 0, 99}); err != ErrNotRegistered {
 		t.Errorf("got %v != exp ErrNotRegistered", err)
 	}
 }
+
+func TestConfluentHeader(t *testing.T) {
+	var h confluentHeader
+
+	for i, test := range []struct {
+		id     int
+		index  []int
+		expEnc []byte
+	}{
+		{id: 1, index: nil, expEnc: []byte{0, 0, 0, 0, 1}},
+		{id: 256, index: nil, expEnc: []byte{0, 0, 0, 1, 0}},
+		{id: 2, index: []int{0}, expEnc: []byte{0, 0, 0, 0, 2, 0}},
+		{id: 3, index: []int{1}, expEnc: []byte{0, 0, 0, 0, 3, 2, 2}},
+		{id: 4, index: []int{1, 2, 3}, expEnc: []byte{0, 0, 0, 0, 4, 6, 2, 4, 6}},
+	} {
+		b, err := h.AppendEncode(nil, test.id, test.index)
+		if err != nil {
+			t.Errorf("#%d AppendEncode: got unexpected err %v", i, err)
+			continue
+		}
+		if !bytes.Equal(b, test.expEnc) {
+			t.Errorf("#%d: AppendEncode(%v) != exp(%v)", i, b, test.expEnc)
+			continue
+		}
+
+		if b2, _ := h.AppendEncode([]byte("foo"), test.id, test.index); !bytes.Equal(b2, append([]byte("foo"), b...)) {
+			t.Errorf("#%d got AppendEncode(%v) != AppendEncode(foo%v)", i, b2, b)
+		}
+
+		id, b2, err := h.DecodeID(b)
+		if err != nil {
+			t.Errorf("#%d DecodeID: got unexpected err %v", i, err)
+			continue
+		}
+		if id != test.id {
+			t.Errorf("#%d: DecodeID: id(%v) != exp(%v)", i, id, test.id)
+			continue
+		}
+		if test.index == nil && len(b2) != 0 {
+			t.Errorf("#%d: DecodeID: bytes(%v) != exp([])", i, b2)
+			continue
+		}
+
+		if test.index != nil {
+			index, b3, err := h.DecodeIndex(b2, len(test.index))
+			if err != nil {
+				t.Errorf("#%d DecodeIndex: got unexpected err %v", i, err)
+				continue
+			}
+			if !reflect.DeepEqual(index, test.index) {
+				t.Errorf("#%d: DecodeIndex: index(%v) != exp(%v)", i, index, test.index)
+				continue
+			}
+			if len(b3) != 0 {
+				t.Errorf("#%d: DecodeIndex: bytes(%v) != exp([])", i, b3)
+				continue
+			}
+		}
+	}
+
+	if _, _, err := h.DecodeID([]byte{1, 0, 0, 0, 0, 1}); err != ErrBadHeader {
+		t.Errorf("got %v != exp ErrBadHeader", err)
+	}
+	if _, _, err := h.DecodeID([]byte{0, 0, 0, 0}); err != ErrBadHeader {
+		t.Errorf("got %v != exp ErrBadHeader", err)
+	}
+	if _, _, err := h.DecodeIndex([]byte{2}, 1); err != io.EOF {
+		t.Errorf("got %v != exp io.EOF", err)
+	}
+	if _, _, err := h.DecodeIndex([]byte{6, 2, 4, 6}, 2); err != ErrNotRegistered {
+		t.Errorf("got %v != exp ErrNotRegistered", err)
+	}
+	if _, _, err := h.DecodeIndex([]byte{1}, 2); err != ErrBadHeader {
+		t.Errorf("got %v != exp ErrBadHeader", err)
+	}
+}