Repository: twmb/franz-go
Pull request #978 (2025-04-17 23:15:48)

sr: refactor DialTLSConfig option so it is stored in client
This PR refactors the `DialTLSConfig` option setter so the TLS config is stored in `sr.Client` and applied in `sr.NewClient` instead of the option setter.

Note that using `DialTLSConfig()` overrides any client set with `HTTPClient()`. This is the same behaviour as before.

After this PR is merged, I will add the missing `OptValues` case for the `DialTLSConfig` option to #963.

Fixes #971 


twmb: In kgo: if DialTLSConfig and Dialer are both present, an error is returned.

In this package, the old code kept whatever option was used last: if you had DialTLSConfig before HTTPClient, the final HTTPClient would be kept -- and vise versa. The new code in this PR unconditionally keeps DialTLSConfig.

I added a comment, wondering what you think, that allows people to use a custom client while opting into DialTLSConfig.
hhromic: > In this package, the old code kept whatever option was used last: if you had DialTLSConfig before HTTPClient, the final HTTPClient would be kept -- and vise versa. The new code in this PR unconditionally keeps DialTLSConfig.

Ah you are right, I misread the existing `NewClient` code. Indeed the optoin order determines precedence.

Thanks for the feedback! Appreciated.
I am okay with the comment you left and left a another question for you.

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index 9b7bac0c..1522c185 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -1,6 +1,6 @@
 version: "2"
 run:
-  go: "1.23"
+  go: "1.22"
   allow-parallel-runners: true
 linters:
   default: none
diff --git a/README.md b/README.md
index f5239b39..98528981 100644
--- a/README.md
+++ b/README.md
@@ -279,7 +279,7 @@ generation.
 | [KIP-35](https://cwiki.apache.org/confluence/display/KAFKA/KIP-35+-+Retrieving+protocol+version) — ApiVersion | 0.10.0 | Supported |
 | [KIP-40](https://cwiki.apache.org/confluence/display/KAFKA/KIP-40%3A+ListGroups+and+DescribeGroup) — ListGroups and DescribeGroups | 0.9.0 | Supported |
 | [KIP-41](https://cwiki.apache.org/confluence/display/KAFKA/KIP-41%3A+KafkaConsumer+Max+Records) — max.poll.records | 0.10.0 | Supported (via PollRecords) |
-| [KIP-42](https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors) — Producer & consumer interceptors | 0.10.0 | Supported via hooks |
+| [KIP-42](https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors) — Producer & consumer interceptors | 0.10.0 | Partial support (hooks) |
 | [KIP-43](https://cwiki.apache.org/confluence/display/KAFKA/KIP-43%3A+Kafka+SASL+enhancements) — SASL PLAIN & handshake | 0.10.0 | Supported |
 | [KIP-48](https://cwiki.apache.org/confluence/display/KAFKA/KIP-48+Delegation+token+support+for+Kafka) — Delegation tokens | 1.1 | Supported |
 | [KIP-54](https://cwiki.apache.org/confluence/display/KAFKA/KIP-54+-+Sticky+Partition+Assignment+Strategy) — Sticky partitioning | 0.11.0 | Supported |
@@ -289,7 +289,7 @@ generation.
 | [KIP-74](https://cwiki.apache.org/confluence/display/KAFKA/KIP-74%3A+Add+Fetch+Response+Size+Limit+in+Bytes) — Fetch response size limits | 0.10.1 | Supported |
 | [KIP-78](https://cwiki.apache.org/confluence/display/KAFKA/KIP-78%3A+Cluster+Id) — ClusterID in Metadata | 0.10.1 | Supported |
 | [KIP-79](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65868090) — List offsets for times | 0.10.1 | Supported |
-| [KIP-81](https://cwiki.apache.org/confluence/display/KAFKA/KIP-81%3A+Bound+Fetch+memory+usage+in+the+consumer) — Bound fetch memory usage | WIP | Supported (through options) |
+| [KIP-81](https://cwiki.apache.org/confluence/display/KAFKA/KIP-81%3A+Bound+Fetch+memory+usage+in+the+consumer) — Bound fetch memory usage | WIP | Supported (through a combo of options) |
 | [KIP-82](https://cwiki.apache.org/confluence/display/KAFKA/KIP-82+-+Add+Record+Headers) — Record headers | 0.11.0 | Supported |
 | [KIP-84](https://cwiki.apache.org/confluence/display/KAFKA/KIP-84%3A+Support+SASL+SCRAM+mechanisms) — SASL SCRAM | 0.10.2 | Supported |
 | [KIP-86](https://cwiki.apache.org/confluence/display/KAFKA/KIP-86%3A+Configurable+SASL+callback+handlers) — SASL Callbacks | 0.10.2 | Supported (through callback fns) |
@@ -304,7 +304,7 @@ generation.
 | [KIP-110](https://cwiki.apache.org/confluence/display/KAFKA/KIP-110%3A+Add+Codec+for+ZStandard+Compression) — zstd | 2.1 | Supported |
 | [KIP-112](https://cwiki.apache.org/confluence/display/KAFKA/KIP-112%3A+Handle+disk+failure+for+JBOD) — Broker request protocol changes | 1.0 | Supported |
 | [KIP-113](https://cwiki.apache.org/confluence/display/KAFKA/KIP-113%3A+Support+replicas+movement+between+log+directories) — LogDir requests | 1.0 | Supported |
-| [KIP-117](https://cwiki.apache.org/confluence/display/KAFKA/KIP-117%3A+Add+a+public+AdminClient+API+for+Kafka+admin+operations) — Admin client | 0.11.0 | Supported |
+| [KIP-117](https://cwiki.apache.org/confluence/display/KAFKA/KIP-117%3A+Add+a+public+AdminClient+API+for+Kafka+admin+operations) — Admin client | 0.11.0 | Supported (via kmsg) |
 | [KIP-124](https://cwiki.apache.org/confluence/display/KAFKA/KIP-124+-+Request+rate+quotas) — Request rate quotas | 0.11.0 | Supported |
 | [KIP-126](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=68715855) — Ensure proper batch size after compression | 0.11.0 | Supported (avoided entirely) |
 | [KIP-133](https://cwiki.apache.org/confluence/display/KAFKA/KIP-133%3A+Describe+and+Alter+Configs+Admin+APIs) — Describe & Alter configs | 0.11.0 | Supported |
@@ -362,7 +362,7 @@ generation.
 | [KIP-498](https://cwiki.apache.org/confluence/display/KAFKA/KIP-498%3A+Add+client-side+configuration+for+maximum+response+size+to+protect+against+OOM) — Max bound on reads | ? | Supported |
 | [KIP-511](https://cwiki.apache.org/confluence/display/KAFKA/KIP-511%3A+Collect+and+Expose+Client%27s+Name+and+Version+in+the+Brokers) — Client name/version in ApiVersions request | 2.4 | Supported |
 | [KIP-514](https://cwiki.apache.org/confluence/display/KAFKA/KIP-514%3A+Add+a+bounded+flush%28%29+API+to+Kafka+Producer) — Bounded Flush | 2.4 | Supported (via context) |
-| [KIP-516](https://cwiki.apache.org/confluence/display/KAFKA/KIP-516%3A+Topic+Identifiers) — Topic IDs | 2.8 | Supported |
+| [KIP-516](https://cwiki.apache.org/confluence/display/KAFKA/KIP-516%3A+Topic+Identifiers) — Topic IDs | ??? | Supported as it is implemented |
 | [KIP-518](https://cwiki.apache.org/confluence/display/KAFKA/KIP-518%3A+Allow+listing+consumer+groups+per+state) — List groups by state | 2.6 | Supported |
 | [KIP-519](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=128650952) — Configurable SSL "engine" | 2.6 | Supported (via dialer) |
 | [KIP-525](https://cwiki.apache.org/confluence/display/KAFKA/KIP-525+-+Return+topic+metadata+and+configs+in+CreateTopics+response) — CreateTopics v5 returns configs | 2.4 | Supported |
@@ -389,7 +389,6 @@ generation.
 | [KIP-700](https://cwiki.apache.org/confluence/display/KAFKA/KIP-700%3A+Add+Describe+Cluster+API) — DescribeCluster | 2.8 | Supported |
 | [KIP-704](https://cwiki.apache.org/confluence/display/KAFKA/KIP-704%3A+Send+a+hint+to+the+partition+leader+to+recover+the+partition) — AlterISR => AlterPartition | 3.2 | Supported |
 | [KIP-709](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=173084258) — Batch OffsetFetch | 3.0 | Supported |
-| [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) - Client Metrics | 3.7 | Supported |
 | [KIP-730](https://cwiki.apache.org/confluence/display/KAFKA/KIP-730%3A+Producer+ID+generation+in+KRaft+mode) - AllocateProducerIDs | 3.0 | Supported |
 | [KIP-734](https://cwiki.apache.org/confluence/display/KAFKA/KIP-734:+Improve+AdminClient.listOffsets+to+return+timestamp+and+offset+for+the+record+with+the+largest+timestamp) — Support MaxTimestamp in ListOffsets | 3.0 | Supported (simple version bump) |
 | [KIP-735](https://cwiki.apache.org/confluence/display/KAFKA/KIP-735%3A+Increase+default+consumer+session+timeout) — Bump default session timeout | ? | Supported |
@@ -402,7 +401,7 @@ generation.
 | [KIP-827](https://cwiki.apache.org/confluence/display/KAFKA/KIP-827%3A+Expose+logdirs+total+and+usable+space+via+Kafka+API) — `DescribeLogDirs.{Total,Usable}Bytes` | 3.3 | Supported |
 | [KIP-836](https://cwiki.apache.org/confluence/display/KAFKA/KIP-836%3A+Addition+of+Information+in+DescribeQuorumResponse+about+Voter+Lag) — `DescribeQuorum` voter lag info | 3.3 | Supported |
 | [KIP-841](https://cwiki.apache.org/confluence/display/KAFKA/KIP-841%3A+Fenced+replicas+should+not+be+allowed+to+join+the+ISR+in+KRaft) — `AlterPartition.TopicID` | 3.3 | Supported |
-| [KIP-848](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol) — Next gen consumer rebalance protocol | 3.7 | Supported |
+| [KIP-848](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol) — Next gen consumer rebalance protocol | 3.7 | Unsupported (proto supported) |
 | [KIP-853](https://cwiki.apache.org/confluence/display/KAFKA/KIP-853%3A+KRaft+Controller+Membership+Changes) — Add replica directory ID for replica fetchers | 3.9 | Supported |
 | [KIP-858](https://cwiki.apache.org/confluence/display/KAFKA/KIP-858%3A+Handle+JBOD+broker+disk+failure+in+KRaft) — JBOD in KRaft (protocol) | 3.7 | Supported |
 | [KIP-866](https://cwiki.apache.org/confluence/display/KAFKA/KIP-866+ZooKeeper+to+KRaft+Migration) — ZK to Raft RPC changes | 3.4 | Supported |
@@ -416,11 +415,9 @@ generation.
 | [KIP-994](https://cwiki.apache.org/confluence/display/KAFKA/KIP-994%3A+Minor+Enhancements+to+ListTransactions+and+DescribeTransactions+APIs) — List/Describe transactions enhancements | 3.8 (partial) | Supported |
 | [KIP-1000](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1000%3A+List+Client+Metrics+Configuration+Resources) — ListClientMetricsResources | 3.7 | Supported |
 | [KIP-1005](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1005%3A+Expose+EarliestLocalOffset+and+TieredOffset) — ListOffsets w. Timestamp -5 | 3.9 | Supported |
-| [KIP-1106](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1106%3A+Add+duration+based+offset+reset+option+for+consumer+clients) - Reset offset by duration | 4.0 | Skipped, unneeded |
 | [KIP-1043](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1043%3A+Administration+of+groups) — Administration of groups (protocol) | 4.0 | Supported |
 | [KIP-1073](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1073:+Return+fenced+brokers+in+DescribeCluster+response) — DescribeCluster.IsFenced | 4.0 | Supported |
 | [KIP-1075](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1075%3A+Introduce+delayed+remote+list+offsets+purgatory+to+make+LIST_OFFSETS+async) — TimeoutMillis on ListOffsets | 4.0 | Supported |
-| [KIP-1076](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1076%3A++Metrics+for+client+applications+KIP-714+extension) — User provided client metrics | 4.0 | Supported |
 | [KIP-1102](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1102%3A+Enable+clients+to+rebootstrap+based+on+timeout+or+error+code) — RebootstrapRequired | 4.0 | Supported |
 
 Missing from above but included in librdkafka is:
diff --git a/go.mod b/go.mod
index 10a6b1c7..27ee637c 100644
--- a/go.mod
+++ b/go.mod
@@ -1,11 +1,11 @@
 module github.com/twmb/franz-go
 
-go 1.23
+go 1.21
 
 require (
 	github.com/klauspost/compress v1.17.11
 	github.com/pierrec/lz4/v4 v4.1.22
-	github.com/twmb/franz-go/pkg/kmsg v1.11.2
+	github.com/twmb/franz-go/pkg/kmsg v1.11.0
 	golang.org/x/crypto v0.32.0
 )
 
diff --git a/go.sum b/go.sum
index c4a5de91..219a04b9 100644
--- a/go.sum
+++ b/go.sum
@@ -2,7 +2,7 @@ github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IX
 github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
 github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
-github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg=
-github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE=
+github.com/twmb/franz-go/pkg/kmsg v1.11.0 h1:zUY4I4xgNhQh3M0tPPWkJbfGBvsW9fM1XyIkAaB7AZ0=
+github.com/twmb/franz-go/pkg/kmsg v1.11.0/go.mod h1:CMbfazviCyY6HM0SXuG5t9vOwYDHRCSrJJyBAe5paqg=
 golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
 golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
diff --git a/pkg/kfake/01_fetch.go b/pkg/kfake/01_fetch.go
index f69a84f2..a674e7c6 100644
--- a/pkg/kfake/01_fetch.go
+++ b/pkg/kfake/01_fetch.go
@@ -183,10 +183,10 @@ full:
 			}
 			var pbytes int
 			for _, b := range pd.batches[i:] {
-				if nbytes = nbytes + b.nbytes; nbytes > int(req.MaxBytes) && batchesAdded > 0 {
+				if nbytes = nbytes + b.nbytes; nbytes > int(req.MaxBytes) && batchesAdded > 1 {
 					break full
 				}
-				if pbytes = pbytes + b.nbytes; pbytes > int(rp.PartitionMaxBytes) && batchesAdded > 0 {
+				if pbytes = pbytes + b.nbytes; pbytes > int(rp.PartitionMaxBytes) && batchesAdded > 1 {
 					break
 				}
 				batchesAdded++
diff --git a/pkg/kfake/18_api_versions.go b/pkg/kfake/18_api_versions.go
index 39a4a85f..c4378da4 100644
--- a/pkg/kfake/18_api_versions.go
+++ b/pkg/kfake/18_api_versions.go
@@ -5,23 +5,22 @@ import (
 	"sort"
 	"sync"
 
-	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kmsg"
 )
 
-func init() { regKey(18, 0, 4) }
+func init() { regKey(18, 0, 3) }
 
 func (c *Cluster) handleApiVersions(kreq kmsg.Request) (kmsg.Response, error) {
 	req := kreq.(*kmsg.ApiVersionsRequest)
 	resp := req.ResponseKind().(*kmsg.ApiVersionsResponse)
 
-	if resp.Version > 3 && resp.Version > apiVersionsKeys[18].MaxVersion {
+	if resp.Version > 3 {
 		resp.Version = 0 // downgrades to 0 if the version is unknown
-		resp.ErrorCode = kerr.UnsupportedVersion.Code
 	}
 
-	// We do not checkReqVersion for ApiVersions; if the client uses a
-	// version larger than we support, we auto-downgrade.
+	if err := checkReqVersion(req.Key(), req.Version); err != nil {
+		return nil, err
+	}
 
 	// If we are handling ApiVersions, our package is initialized and we
 	// build our response once.
diff --git a/pkg/kfake/go.mod b/pkg/kfake/go.mod
index f9a224c0..fd073c16 100644
--- a/pkg/kfake/go.mod
+++ b/pkg/kfake/go.mod
@@ -1,6 +1,6 @@
 module github.com/twmb/franz-go/pkg/kfake
 
-go 1.23
+go 1.21
 
 toolchain go1.22.0
 
diff --git a/pkg/kgo/broker.go b/pkg/kgo/broker.go
index e69a1ef1..d4146add 100644
--- a/pkg/kgo/broker.go
+++ b/pkg/kgo/broker.go
@@ -41,8 +41,6 @@ func (p *pinReq) SetVersion(v int16) {
 	p.Request.SetVersion(v)
 }
 
-type forceOpenReq struct{ kmsg.Request }
-
 type promisedReq struct {
 	ctx     context.Context
 	req     kmsg.Request
@@ -410,16 +408,6 @@ start:
 	default:
 	}
 
-	if _, isForceOpen := req.(*forceOpenReq); isForceOpen {
-		// We issue ApiVersions with v0; we could try to bound the
-		// version by going to the start above, but it really does
-		// not matter much.
-		kreq := kmsg.NewPtrApiVersionsRequest()
-		kreq.ClientSoftwareName = b.cl.cfg.softwareName
-		kreq.ClientSoftwareVersion = b.cl.cfg.softwareVersion
-		req = kreq
-	}
-
 	// Produce requests (and only produce requests) can be written
 	// without receiving a reply. If we see required acks is 0,
 	// then we immediately call the promise with no response.
@@ -511,8 +499,7 @@ func (p bufPool) put(b []byte) { p.p.Put(&b) }
 func (b *broker) loadConnection(ctx context.Context, req kmsg.Request) (*brokerCxn, error) {
 	var (
 		pcxn         = &b.cxnNormal
-		isProduceCxn bool
-		isFetchCxn   bool
+		isProduceCxn bool // see docs on brokerCxn.discard for why we do this
 		reqKey       = req.Key()
 		_, isTimeout = req.(kmsg.TimeoutRequest)
 	)
@@ -522,7 +509,6 @@ func (b *broker) loadConnection(ctx context.Context, req kmsg.Request) (*brokerC
 		isProduceCxn = true
 	case reqKey == 1:
 		pcxn = &b.cxnFetch
-		isFetchCxn = true
 	case reqKey == 11 || reqKey == 14: // join || sync
 		pcxn = &b.cxnGroup
 	case isTimeout:
@@ -553,12 +539,6 @@ func (b *broker) loadConnection(ctx context.Context, req kmsg.Request) (*brokerC
 	}
 	b.cl.cfg.logger.Log(LogLevelDebug, "connection initialized successfully", "addr", b.addr, "broker", logID(b.meta.NodeID))
 
-	if isProduceCxn {
-		b.cl.metrics.observeRate(&b.cl.metrics.pConnCreation)
-	} else if isFetchCxn {
-		b.cl.metrics.observeRate(&b.cl.metrics.cConnCreation)
-	}
-
 	b.reapMu.Lock()
 	defer b.reapMu.Unlock()
 	*pcxn = cxn
@@ -786,7 +766,6 @@ start:
 			maxVersion = 0
 			goto start
 		}
-		cxn.cl.cfg.logger.Log(LogLevelDebug, "broker does not know our ApiVersions version but replied with all keys, deserializing as v0", "broker", logID(cxn.b.meta.NodeID))
 		resp.Version = 0
 	}
 
@@ -1253,16 +1232,6 @@ func (cxn *brokerCxn) readResponse(
 			})
 		}
 	})
-
-	if readErr == nil {
-		latencyMillis := (writeWait + timeToWrite + readWait + timeToRead).Milliseconds()
-		if key == 0 { //nolint:staticcheck // tagged switch not needed for one case...
-			cxn.b.cl.metrics.observeNodeTime(cxn.b.meta.NodeID, &cxn.b.cl.metrics.pReqLatency, latencyMillis)
-		} else if key == 1 {
-			cxn.b.cl.metrics.observeNodeTime(cxn.b.meta.NodeID, &cxn.b.cl.metrics.cReqLatency, latencyMillis)
-		}
-	}
-
 	if logger := cxn.cl.cfg.logger; logger.Level() >= LogLevelDebug {
 		logger.Log(LogLevelDebug, fmt.Sprintf("read %s v%d", kmsg.NameForKey(key), version), "broker", logID(cxn.b.meta.NodeID), "bytes_read", bytesRead, "read_wait", readWait, "time_to_read", timeToRead, "err", readErr)
 	}
@@ -1504,11 +1473,6 @@ func (cxn *brokerCxn) handleResp(pr promisedResp) {
 		return
 	}
 
-	if pr.resp.Key() == 18 && len(rawResp) > 2 && rawResp[1] == 35 {
-		cxn.b.cl.cfg.logger.Log(LogLevelDebug, "ApiVersions response returned with UNSUPPORTED_VERSION error at byte 1, downgraded to v0 to deserialize response")
-		pr.resp.SetVersion(0)
-	}
-
 	cxn.successes++
 	readErr := pr.resp.ReadFrom(rawResp)
 
@@ -1520,9 +1484,6 @@ func (cxn *brokerCxn) handleResp(pr promisedResp) {
 		if throttleResponse, ok := pr.resp.(kmsg.ThrottleResponse); ok {
 			millis, throttlesAfterResp := throttleResponse.Throttle()
 			if millis > 0 {
-				if pr.resp.Key() == 0 {
-					cxn.b.cl.metrics.observeTime(&cxn.b.cl.metrics.pThrottle, int64(millis))
-				}
 				cxn.b.cl.cfg.logger.Log(LogLevelInfo, "broker is throttling us in response", "broker", logID(cxn.b.meta.NodeID), "req", kmsg.Key(pr.resp.Key()).Name(), "throttle_millis", millis, "throttles_after_resp", throttlesAfterResp)
 				if throttlesAfterResp {
 					throttleUntil := time.Now().Add(time.Millisecond * time.Duration(millis)).UnixNano()
diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go
index 74caa84c..4053dc05 100644
--- a/pkg/kgo/client.go
+++ b/pkg/kgo/client.go
@@ -81,8 +81,6 @@ type Client struct {
 	consumer consumer
 	id2t     atomic.Value // map[[16]byte]string
 
-	metrics metrics
-
 	coordinatorsMu sync.Mutex
 	coordinators   map[coordinatorKey]*coordinatorLoad
 
@@ -259,8 +257,6 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.dialTLS}
 	case namefn(DialTLS):
 		return []any{cfg.dialTLS != nil}
-	case namefn(DialTimeout):
-		return []any{cfg.dialTimeout}
 	case namefn(SeedBrokers):
 		return []any{cfg.seedBrokers}
 	case namefn(MaxVersions):
@@ -297,12 +293,6 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.missingTopicDelete}
 	case namefn(OnRebootstrapRequired):
 		return []any{cfg.onRebootstrapRequired}
-	case namefn(WithContext):
-		return []any{cfg.ctx}
-	case namefn(DisableClientMetrics):
-		return []any{cfg.disableClientMetrics}
-	case namefn(UserMetricsFn):
-		return []any{cfg.userMetrics}
 
 	case namefn(DefaultProduceTopic):
 		return []any{cfg.defaultProduceTopic}
@@ -358,8 +348,6 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.decompressor}
 	case namefn(ConsumeRegex):
 		return []any{cfg.regex}
-	case namefn(ConsumeStartOffset):
-		return []any{cfg.startOffset}
 	case namefn(ConsumeResetOffset):
 		return []any{cfg.resetOffset}
 	case namefn(ConsumeTopics):
@@ -430,7 +418,8 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.requireStable}
 	case namefn(SessionTimeout):
 		return []any{cfg.sessionTimeout}
-
+	case namefn(WithContext):
+		return []any{cfg.ctx}
 	default:
 		return nil
 	}
@@ -488,12 +477,6 @@ func NewClient(opts ...Opt) (*Client, error) {
 		}
 	}
 
-	if cfg.setResetOffset && !cfg.setStartOffset {
-		cfg.startOffset = cfg.resetOffset
-	} else if cfg.setStartOffset && !cfg.setResetOffset {
-		cfg.resetOffset = cfg.startOffset
-	} // else they are both set (keep) or both unset (defaults)
-
 	ctx := context.Background()
 
 	if cfg.ctx != nil {
@@ -547,7 +530,6 @@ func NewClient(opts ...Opt) (*Client, error) {
 	cl.producer.init(cl)
 	cl.consumer.init(cl)
 	cl.metawait.init()
-	cl.metrics.init(cl)
 
 	if cfg.id != nil {
 		cl.reqFormatter = kmsg.NewRequestFormatter(kmsg.FormatterClientID(*cfg.id))
@@ -561,7 +543,6 @@ func NewClient(opts ...Opt) (*Client, error) {
 	cl.seeds.Store(seedBrokers)
 	go cl.updateMetadataLoop()
 	go cl.reapConnectionsLoop()
-	go cl.pushMetrics()
 
 	return cl, nil
 }
@@ -586,14 +567,14 @@ func (cl *Client) loadSeeds() []*broker {
 	return cl.seeds.Load().([]*broker)
 }
 
-// Ping returns whether any broker is reachable and that the client can
-// communicate with it, iterating over any discovered broker or seed broker
-// until one returns a successful response to a broker-only Metadata request.
-// No discovered broker nor seed broker is attempted more than once. If all
-// requests fail, this returns final error.
+// Ping returns whether any broker is reachable, iterating over any discovered
+// broker or seed broker until one returns a successful response to an
+// ApiVersions request. No discovered broker nor seed broker is attempted more
+// than once. If all requests fail, this returns final error.
 func (cl *Client) Ping(ctx context.Context) error {
-	req := kmsg.NewPtrMetadataRequest()
-	req.Topics = []kmsg.MetadataRequestTopic{}
+	req := kmsg.NewPtrApiVersionsRequest()
+	req.ClientSoftwareName = cl.cfg.softwareName
+	req.ClientSoftwareVersion = cl.cfg.softwareVersion
 
 	cl.brokersMu.RLock()
 	brokers := append([]*broker(nil), cl.brokers...)
@@ -605,9 +586,8 @@ func (cl *Client) Ping(ctx context.Context) error {
 		cl.loadSeeds(),
 	} {
 		for _, br := range brs {
-			resp, err := br.waitResp(ctx, req)
+			_, err := br.waitResp(ctx, req)
 			if lastErr = err; lastErr == nil {
-				cl.updateMetadataBrokers(resp.(*kmsg.MetadataResponse))
 				return nil
 			}
 		}
@@ -881,12 +861,6 @@ func (cl *Client) supportsKIP848v1() bool {
 	return cl.supportsKeyVersion(int16(kmsg.ConsumerGroupHeartbeat), 1)
 }
 
-// Called after the first metric observed, which is always after a response.
-func (cl *Client) supportsClientMetrics() bool {
-	return cl.supportsKeyVersion(int16(kmsg.GetTelemetrySubscriptions), 0) &&
-		cl.supportsKeyVersion(int16(kmsg.PushTelemetry), 0)
-}
-
 // A broker may not support some requests we want to make. This function checks
 // support. This should only be used *after* at least one successful response.
 func (cl *Client) supportsKeyVersion(key, version int16) bool {
@@ -929,9 +903,7 @@ func (cl *Client) fetchBrokerMetadata(ctx context.Context) error {
 		close(wait.done)
 	}()
 
-	req := kmsg.NewPtrMetadataRequest()
-	req.Topics = []kmsg.MetadataRequestTopic{}
-	_, _, wait.err = cl.fetchMetadata(ctx, req, true, nil)
+	_, _, wait.err = cl.fetchMetadata(ctx, kmsg.NewPtrMetadataRequest(), true, nil)
 	return wait.err
 }
 
@@ -984,7 +956,13 @@ start:
 				}
 			}
 		}
-		cl.updateMetadataBrokers(meta)
+		cl.controllerIDMu.Lock()
+		if meta.ControllerID >= 0 {
+			cl.controllerID = meta.ControllerID
+		}
+		cl.clusterID = meta.ClusterID
+		cl.controllerIDMu.Unlock()
+		cl.updateBrokers(meta.Brokers)
 
 		// Cache the mapped metadata, and potentially store each topic in the results.
 		cl.storeCachedMappedMetadata(meta, intoMapped)
@@ -992,16 +970,6 @@ start:
 	return r.last, meta, err
 }
 
-func (cl *Client) updateMetadataBrokers(resp *kmsg.MetadataResponse) {
-	cl.controllerIDMu.Lock()
-	if resp.ControllerID >= 0 {
-		cl.controllerID = resp.ControllerID
-	}
-	cl.clusterID = resp.ClusterID
-	cl.controllerIDMu.Unlock()
-	cl.updateBrokers(resp.Brokers)
-}
-
 // updateBrokers is called with the broker portion of every metadata response.
 // All metadata responses contain all known live brokers, so we can always
 // use the response.
@@ -1133,21 +1101,8 @@ func (cl *Client) close(ctx context.Context) (rerr error) {
 
 	// Now we kill the client context and all brokers, ensuring all
 	// requests fail. This will finish all producer callbacks and
-	// stop the metadata loop and metrics loop.
+	// stop the metadata loop.
 	cl.ctxCancel()
-
-	// Before killing brokers, give metrics 1s to push any final
-	// terminating message. The client context cancelation awakens
-	// the push-period-wait loop.
-	after := time.NewTimer(time.Second)
-	select {
-	case <-cl.metrics.ctx.Done():
-	case <-after.C:
-		cl.metrics.ctxCancel()
-	case <-ctx.Done():
-		cl.metrics.ctxCancel()
-	}
-
 	cl.brokersMu.Lock()
 	cl.stopBrokers = true
 	for _, broker := range cl.brokers {
@@ -1558,7 +1513,8 @@ func (cl *Client) shardedRequest(ctx context.Context, req kmsg.Request) ([]Respo
 		return shards(cl.handleAdminReq(ctx, t)), nil
 
 	case kmsg.GroupCoordinatorRequest,
-		kmsg.TxnCoordinatorRequest:
+		kmsg.TxnCoordinatorRequest,
+		*kmsg.ConsumerGroupHeartbeatRequest:
 		return shards(cl.handleCoordinatorReq(ctx, t)), nil
 
 	case *kmsg.ApiVersionsRequest:
@@ -2047,15 +2003,8 @@ func (cl *Client) handleCoordinatorReq(ctx context.Context, req kmsg.Request) Re
 		return cl.handleCoordinatorReqSimple(ctx, coordinatorTypeGroup, t.Group, req)
 	case *kmsg.OffsetDeleteRequest:
 		return cl.handleCoordinatorReqSimple(ctx, coordinatorTypeGroup, t.Group, req)
-
-	// ConsumerGroupHeartbeat cannot be retried at all
 	case *kmsg.ConsumerGroupHeartbeatRequest:
-		br, err := cl.loadCoordinator(ctx, coordinatorTypeGroup, t.Group)
-		var resp kmsg.Response
-		if err == nil {
-			resp, err = br.waitResp(ctx, req)
-		}
-		return shard(br, req, resp, err)
+		return cl.handleCoordinatorReqSimple(ctx, coordinatorTypeGroup, t.Group, req)
 	}
 }
 
diff --git a/pkg/kgo/compression.go b/pkg/kgo/compression.go
index 508ea9df..3e07bb1d 100644
--- a/pkg/kgo/compression.go
+++ b/pkg/kgo/compression.go
@@ -65,25 +65,6 @@ func Lz4Compression() CompressionCodec { return CompressionCodec{CodecLz4, 0} }
 // ZstdCompression enables zstd compression with the default compression level.
 func ZstdCompression() CompressionCodec { return CompressionCodec{CodecZstd, 0} }
 
-// CompressFlag is a flag to instruct the compressor.
-type CompressFlag uint16
-
-const (
-	// CompressDisableZstd instructs the compressor that zstd should not be
-	// used, even if the compressor supports it. This is used when
-	// producing to an old broker (pre Kafka v2.1) that does not yet
-	// support zstd compression. If you are confident you will only produce
-	// to new brokers, you can ignore this flag.
-	CompressDisableZstd CompressFlag = 1 + iota
-)
-
-func mkCompressFlags(produceRequestVersion int16) []CompressFlag {
-	if produceRequestVersion < 7 {
-		return []CompressFlag{CompressDisableZstd}
-	}
-	return nil
-}
-
 // Compressor is an interface that defines how produce batches are compressed.
 // You can override the default client internal compressor for more control
 // over what compressors to use, level, and memory reuse.
@@ -96,11 +77,10 @@ type Compressor interface {
 	// pool. As an example, you can look at the franz-go internal
 	// implementation of the default compressor in compression.go.
 	//
-	// Flags may optionally be provided to direct the compressor to enable
-	// or disable features. New backwards compatible flags may be
-	// introduced. If you add features to your compressor, be sure to
-	// evaluate if new flags exist to opt into or out of features.
-	Compress(dst *bytes.Buffer, src []byte, flags ...CompressFlag) ([]byte, CompressionCodecType)
+	// Note that old brokers do not support zstd compression. If the
+	// produce request version is less than 7, zstd compression should
+	// not be used.
+	Compress(dst *bytes.Buffer, src []byte, produceRequestVersion int16) ([]byte, CompressionCodecType)
 }
 
 // Decompressor is an interface that defines how fetch batches are
@@ -229,17 +209,10 @@ type zstdEncoder struct {
 	inner *zstd.Encoder
 }
 
-func (c *compressor) Compress(dst *bytes.Buffer, src []byte, flags ...CompressFlag) ([]byte, CompressionCodecType) {
-	var disableZstd bool
-	for _, flag := range flags {
-		if flag == CompressDisableZstd {
-			disableZstd = true
-		}
-	}
-
+func (c *compressor) Compress(dst *bytes.Buffer, src []byte, produceRequestVersion int16) ([]byte, CompressionCodecType) {
 	var use CompressionCodecType
 	for _, option := range c.options {
-		if option == CodecZstd && disableZstd {
+		if option == CodecZstd && produceRequestVersion < 7 {
 			continue
 		}
 		use = option
diff --git a/pkg/kgo/config.go b/pkg/kgo/config.go
index 5df777ce..adce37ef 100644
--- a/pkg/kgo/config.go
+++ b/pkg/kgo/config.go
@@ -5,7 +5,6 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
-	"iter"
 	"math"
 	"math/rand"
 	"net"
@@ -102,9 +101,6 @@ type cfg struct {
 
 	sasls []sasl.Mechanism
 
-	disableClientMetrics bool
-	userMetrics          func() iter.Seq[Metric]
-
 	hooks hooks
 	pools pools
 
@@ -147,10 +143,7 @@ type cfg struct {
 	minBytes       int32
 	maxBytes       lazyI32
 	maxPartBytes   lazyI32
-	startOffset    Offset
 	resetOffset    Offset
-	setStartOffset bool
-	setResetOffset bool
 	isolationLevel int8
 	keepControl    bool
 	rack           string
@@ -201,8 +194,6 @@ type cfg struct {
 	autocommitMarks    bool
 	autocommitInterval time.Duration
 	commitCallback     func(*Client, *kmsg.OffsetCommitRequest, *kmsg.OffsetCommitResponse, error)
-
-	disableNextGenBalancer bool
 }
 
 func (cfg *cfg) validate() error {
@@ -563,7 +554,6 @@ func defaultCfg() cfg {
 		minBytes:       1,
 		maxBytes:       50 << 20,
 		maxPartBytes:   1 << 20,
-		startOffset:    NewOffset().AtStart(),
 		resetOffset:    NewOffset().AtStart(),
 		isolationLevel: 0,
 
@@ -950,32 +940,6 @@ func OnRebootstrapRequired(fn func() ([]string, error)) Opt {
 	return clientOpt{func(cfg *cfg) { cfg.onRebootstrapRequired = fn }}
 }
 
-// DisableClientMetrics opts out of collecting and sending client metrics to
-// the broker (if the broker supports receiving client metrics). By default,
-// clients are recommended to gather a small set of metrics to help cluster
-// operators debug client issues (rather than relying on clients which may not
-// be instrumented at all).
-//
-// For more details on client metrics, see KIP-714.
-func DisableClientMetrics() Opt {
-	return clientOpt{func(cfg *cfg) { cfg.disableClientMetrics = true }}
-}
-
-// UserMetricsFn sets the function to call to add user metrics when rolling up
-// client metrics to send to the broker. Every metric rollup, fn is called and
-// returns an iterator. All metrics returned from the iterator are included in
-// the client metric aggregation and are sent to the broker. It is your
-// responsibility to ensure the metric name is formatted correctly (namespaced
-// and following OpenTelemetry format), and you need to ensure your Sum metrics
-// are monotonically increasing. See the documentation on [Metric] for more
-// details.
-//
-// For more details about the client sending metrics, see KIP-714. For more
-// details about enhancing client metrics with user metrics, see KIP-1076.
-func UserMetricsFn(fn func() iter.Seq[Metric]) Opt {
-	return clientOpt{func(cfg *cfg) { cfg.userMetrics = fn }}
-}
-
 ////////////////////////////
 // PRODUCER CONFIGURATION //
 ////////////////////////////
@@ -1375,73 +1339,33 @@ func MaxConcurrentFetches(n int) ConsumerOpt {
 	return consumerOpt{func(cfg *cfg) { cfg.maxConcurrentFetches = n }}
 }
 
-// ConsumeStartOffset sets the offset to start consuming from when consuming a
-// partition for the first time. If you do not set [ConsumeResetOffset], this
-// is also the offset to reset to if the client sees an OffsetOutOfRange error
-// while consuming a partition. The default is NewOffset().AtStart(), i.e.,
-// start processing a partition from the earliest offset. If using this option,
-// it is strongly recommended to also set ConsumeResetOffset.
+// ConsumeResetOffset sets the offset to start consuming from, or if
+// OffsetOutOfRange is seen while fetching, to restart consuming from. The
+// default is NewOffset().AtStart(), i.e., the earliest offset.
 //
-// If you use an exact or relative offsets and the offset ends up out of range,
-// the client chooses the nearest of either the log start offset or the log end
-// offset. For example, using At(3) when the partition starts at 8 results in
-// the partition being consumed from offset 8.
+// For direct consumers, this is the offset that partitions begin to consume
+// from. For group consumers, this is the offset that partitions begin to
+// consume from if a partition has no commits. If partitions have commits, the
+// commit offset is used. While fetching, if OffsetOutOfRange is encountered,
+// the partition resets to ConsumeResetOffset. Using [NoResetOffset] stops
+// consuming a partition if the client encounters OffsetOutOfRange. Using
+// [Offset.AtCommitted] prevents consuming a partition in a group if the
+// partition has no prior commits.
 //
-// For group consuming, you can use [Offset.AtCommitted] to prevent starting
-// consuming a partition in a group if the partition has no prior commits.
+// If you use an exact offset or relative offsets and the offset ends up out of
+// range, the client chooses the nearest of either the log start offset or the
+// high watermark: using At(3) when the partition starts at 8 results in the
+// partition being consumed from offset 8.
 //
-// The following determines the offset for when a partition is seen for the
-// first time:
+// In short form, the following determines the offset for when a partition is
+// seen for the first time, or reset while fetching:
 //
-//	at start?                         => start at the log start offset
-//	at end?                           => start at the log end offset
-//	at exact?                         => start at the an exact offset (3 means offset 3)
-//	relative?                         => start at the the above, + / - the relative amount
-//	exact/relative are out of bounds? => start at the nearest boundary (start or end)
-//	after millisec?                   => start at first offset after millisec if one exists, else log end offset
-//
-// To match Kafka's auto.offset.reset which is used for both the start offset
-// and the reset offset,
-//
-//	NewOffset().AtStart()     == auto.offset.reset "earliest"
-//	NewOffset().AtEnd()       == auto.offset.reset "latest"
-//	NewOffset().AtCommitted() == auto.offset.reset "none"
-//
-// Be sure to check the documentation for [ConsumeResetOffset], especially if
-// you rely on this option as the reset offset as well.
-func ConsumeStartOffset(offset Offset) ConsumerOpt {
-	return consumerOpt{func(cfg *cfg) { cfg.startOffset, cfg.setStartOffset = offset, true }}
-}
-
-// ConsumeResetOffset sets the offset to reset to if the client ever sees
-// OffsetOutOfRange while fetching. If you do not set [ConsumeStartOffset],
-// this is also the offset to start consuming from when consuming a partition
-// for the first time. The default is NewOffset().AtStart(), i.e., reset to the
-// earliest offset. If using this option, it is strongly recommended to also
-// set ConsumeStartOffset.
-//
-// This option is *only* used if a consumer seeds OffsetOutOfRange on the
-// *first* fetch of a partition. If the consumer has consumed the partition at
-// all and sees the error, it will automatically reset to the first offset
-// after the timestamp of the last successfully consumed offset. If data loss
-// occurred such that even the last successfully consumed offset is lost, the
-// client automatically resets to the new current end offset. If you want to
-// disable offset resetting entirely, you can use [NoResetOffset].
-//
-// If you use an exact or relative offsets and the offset ends up out of range,
-// the client chooses the nearest of either the log start offset or the log end
-// offset. For example, using At(3) when the partition starts at 8 results in
-// the partition being consumed from offset 8.
-//
-// The following determines the offset for when a partition is seen for the
-// first time, or reset while fetching:
-//
-//	at start?                         => reset to the log start offset
-//	at end?                           => reset to the log end offset
-//	at exact?                         => reset to the an exact offset (3 means offset 3)
-//	relative?                         => reset to the the above, + / - the relative amount
-//	exact/relative are out of bounds? => reset to the nearest boundary (start or end)
-//	after millisec?                   => reset to the first offset after millisec if one exists, else the log end offset
+//	reset at start?                        => log start offset
+//	reset at end?                          => high watermark
+//	reset at exact?                        => this exact offset (3 means offset 3)
+//	reset relative?                        => the above, + / - the relative amount
+//	reset exact or relative out of bounds? => nearest boundary (start or end)
+//	reset after millisec?                  => high watermark, or first offset after millisec if one exists
 //
 // To match Kafka's auto.offset.reset,
 //
@@ -1449,14 +1373,11 @@ func ConsumeStartOffset(offset Offset) ConsumerOpt {
 //	NewOffset().AtEnd()       == auto.offset.reset "latest"
 //	NewOffset().AtCommitted() == auto.offset.reset "none"
 //
-// With the above, make sure to use [NoResetOffset] if you want to stop
+// With the above, make sure to use NoResetOffset() if you want to stop
 // consuming when you encounter OffsetOutOfRange. It is highly recommended
-// to read the docs for all Offset methods.
-//
-// Be sure to check the documentation for [ConsumeStartOffset], especially if
-// you rely on this option as the start offset as well.
+// to read the docs for all Offset methods to see a few other alternatives.
 func ConsumeResetOffset(offset Offset) ConsumerOpt {
-	return consumerOpt{func(cfg *cfg) { cfg.resetOffset, cfg.setResetOffset = offset, true }}
+	return consumerOpt{func(cfg *cfg) { cfg.resetOffset = offset }}
 }
 
 // Rack specifies where the client is physically located and changes fetch
@@ -1973,22 +1894,3 @@ func AutoCommitCallback(fn func(*Client, *kmsg.OffsetCommitRequest, *kmsg.Offset
 		}
 	}}
 }
-
-// !!! Only uncomment once we trust the broker implementation!
-// !!! And add this option to Opt!
-//
-// DisableNextGenRebalancer opts out of the "next gen" rebalancer that is
-// the default as of Kafka 4.0+. The client opts in to the next gen rebalancer
-// automatically if the broker supports it AND if you are using either the
-// [RangeBalancer] or [StickyBalancer] or [CooperativeStickyBalancer]. If you
-// use your own rebalancer or use the [RoundRobinBalancer] or are talking to
-// a broker that does not support the next gen balancer, the client uses the
-// old client-driven group balancing behavior.
-//
-// You may want to use this function if you notice a regression or run into
-// a broker or client bug, or if you prefer the performance of the old
-// client driven rebalancers.
-//  func DisableNextGenRebalancer() GroupOpt {
-//  	return groupOpt{func(cfg *cfg) { cfg.disableNextGenBalancer = true }}
-//  }
-//
diff --git a/pkg/kgo/consumer.go b/pkg/kgo/consumer.go
index 0ee43070..1b84b578 100644
--- a/pkg/kgo/consumer.go
+++ b/pkg/kgo/consumer.go
@@ -147,9 +147,9 @@ func (o Offset) Relative(n int64) Offset {
 	return o
 }
 
-// WithEpoch copies 'o' and returns an offset with the given epoch. This epoch
-// is used for truncation detection; the default of -1 implies no truncation
-// detection.
+// WithEpoch copies 'o' and returns an offset with the given epoch.  to use the
+// given epoch. This epoch is used for truncation detection; the default of -1
+// implies no truncation detection.
 func (o Offset) WithEpoch(e int32) Offset {
 	o.afterMilli = false
 	if e < 0 {
@@ -1109,18 +1109,6 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
 
 	c.cl.cfg.logger.Log(LogLevelDebug, "assign requires loading offsets")
 
-	// We could have a prior handleListOrEpochResults finishing concurrent
-	// with a new assignment. We need to guard below, because list/epoch
-	// results can set offsets for loaded cursors.
-	//
-	// Example: I was just cooperatively assigned p1o10e3, and I issued an
-	// epoch request to validate data loss. I was then assigned p2o4e-1:
-	// there is no epoch to validate, so we immediately begin consuming
-	// at the offset. The epoch request returns and tries also using the
-	// cursor concurrently. We need to guard it.
-	session.listOrEpochMu.Lock()
-	defer session.listOrEpochMu.Unlock()
-
 	topics := tps.load()
 	for topic, partitions := range assignments {
 		topicPartitions := topics.loadTopic(topic) // should be non-nil
@@ -1498,9 +1486,6 @@ type consumerSession struct {
 	workersCond *sync.Cond
 	workers     int
 
-	// listOrEpochMu largely guards the below. It is a sub-mutex of the
-	// consumer mutex to guard one concurrent data access (see below in
-	// assignPartitions).
 	listOrEpochMu           sync.Mutex
 	listOrEpochLoadsWaiting listOrEpochLoads
 	listOrEpochMetaCh       chan struct{} // non-nil if Loads is non-nil, signalled on meta update
@@ -2313,7 +2298,7 @@ func (*Client) loadEpochsForBrokerLoad(ctx context.Context, broker *broker, load
 			offset := loadPart.at
 			var err error
 			if rPartition.EndOffset < offset {
-				err = &ErrDataLoss{topic, partition, offset, loadPart.epoch, rPartition.EndOffset, rPartition.LeaderEpoch}
+				err = &ErrDataLoss{topic, partition, offset, rPartition.EndOffset}
 				offset = rPartition.EndOffset
 			}
 
diff --git a/pkg/kgo/consumer_direct.go b/pkg/kgo/consumer_direct.go
index f81c40f9..9f808a16 100644
--- a/pkg/kgo/consumer_direct.go
+++ b/pkg/kgo/consumer_direct.go
@@ -86,7 +86,7 @@ func (d *directConsumer) findNewAssignments() map[string]map[int32]Offset {
 			}
 			toUseTopic := make(map[int32]Offset, len(partitions.partitions))
 			for partition := range partitions.partitions {
-				toUseTopic[int32(partition)] = d.cfg.startOffset
+				toUseTopic[int32(partition)] = d.cfg.resetOffset
 			}
 			toUse[topic] = toUseTopic
 		}
diff --git a/pkg/kgo/consumer_group.go b/pkg/kgo/consumer_group.go
index 73aaed64..d3904419 100644
--- a/pkg/kgo/consumer_group.go
+++ b/pkg/kgo/consumer_group.go
@@ -163,7 +163,6 @@ type groupConsumer struct {
 	// If we detect we should run in 848 mode, we set is848 true.
 	managing bool
 	is848    bool
-	g848     *g848
 
 	dying    bool // set when closing, read in findNewAssignments
 	left     chan struct{}
@@ -546,7 +545,7 @@ func (g *groupConsumer) leave(ctx context.Context) {
 		// member ID, there's nothing we can do.
 		memberID := g.memberGen.memberID()
 		if memberID == "" {
-			g.cfg.logger.Log(LogLevelInfo, "tried to leave group but we have no member ID yet, returning early", "group", g.cfg.group)
+			g.cfg.logger.Log(LogLevelInfo, "tried to leave group but we have no member ID yet, returning early", g.cfg.group)
 			return
 		}
 
@@ -1603,22 +1602,15 @@ func (g *groupConsumer) fetchOffsets(ctx context.Context, added map[string][]int
 	// request, if we are only requesting one group, as well as maps the
 	// response back, so we do not need to worry about v8+ here.
 start:
-	member, gen := g.memberGen.load()
 	req := kmsg.NewPtrOffsetFetchRequest()
+	req.Group = g.cfg.group
 	req.RequireStable = g.cfg.requireStable
-	reqg := kmsg.NewOffsetFetchRequestGroup()
-	reqg.Group = g.cfg.group
-	if member != "" {
-		reqg.MemberID = &member
-		reqg.MemberEpoch = gen
-	}
 	for topic, partitions := range added {
-		reqTopic := kmsg.NewOffsetFetchRequestGroupTopic()
+		reqTopic := kmsg.NewOffsetFetchRequestTopic()
 		reqTopic.Topic = topic
 		reqTopic.Partitions = partitions
-		reqg.Topics = append(reqg.Topics, reqTopic)
+		req.Topics = append(req.Topics, reqTopic)
 	}
-	req.Groups = append(req.Groups, reqg)
 
 	var resp *kmsg.OffsetFetchResponse
 	var err error
@@ -1681,7 +1673,7 @@ start:
 				offset.epoch = rPartition.LeaderEpoch
 			}
 			if rPartition.Offset == -1 {
-				offset = g.cfg.startOffset
+				offset = g.cfg.resetOffset
 			}
 			topicOffsets[rPartition.Partition] = offset
 		}
@@ -2866,8 +2858,6 @@ func (g *groupConsumer) commit(
 	req.MemberID = memberID
 	req.InstanceID = g.cfg.instanceID
 
-	is848 := g.is848 // safe since we are under g.mu
-
 	if ctx.Done() != nil {
 		go func() {
 			select {
@@ -2913,50 +2903,11 @@ func (g *groupConsumer) commit(
 			}
 		}
 
-		var retries int
-	issue:
-		start := time.Now()
 		resp, err := req.RequestWith(commitCtx, g.cl)
 		if err != nil {
 			onDone(g.cl, req, nil, err)
 			return
 		}
-		g.cl.metrics.observeTime(&g.cl.metrics.cCommitLatency, time.Since(start).Milliseconds())
-
-		// With next gen consumer groups, it is possible for the group
-		// to rebalance again immediately after we discover we need to
-		// revoke. We try up to 3x reloading and resending.
-		if is848 && len(resp.Topics) > 0 && len(resp.Topics[0].Partitions) > 0 {
-			ec := resp.Topics[0].Partitions[0].ErrorCode
-			if kerr.ErrorForCode(ec) == kerr.StaleMemberEpoch && retries < 3 {
-				hbreq := g.g848.mkreq()
-				hbresp, err := hbreq.RequestWith(commitCtx, g.cl)
-				if err != nil {
-					g.cl.cfg.logger.Log(LogLevelInfo, "received STALE_MEMBER_EPOCH while committing; unable to force heartbeat; returning original stale error",
-						"last_generation", generation,
-						"err", err,
-					)
-				} else {
-					if err := kerr.ErrorForCode(hbresp.ErrorCode); err != nil { //nolint:revive // err != nil is more standard as the first case
-						g.cl.cfg.logger.Log(LogLevelInfo, "received STALE_MEMBER_EPOCH while committing; forced immediate heartbeat to load the latest generation but received an error",
-							"last_generation", generation,
-							"err", err,
-						)
-					} else {
-						g.cl.cfg.logger.Log(LogLevelInfo, "received STALE_MEMBER_EPOCH while committing; forced immediate heartbeat to load the latest generation",
-							"last_generation", generation,
-							"new_generation", hbresp.MemberEpoch,
-						)
-						g.memberGen.storeGeneration(hbresp.MemberEpoch)
-						generation = hbresp.MemberEpoch
-						req.Generation = generation
-						retries++
-						goto issue
-					}
-				}
-			}
-		}
-
 		g.updateCommitted(req, resp)
 		onDone(g.cl, req, resp, nil)
 	}()
diff --git a/pkg/kgo/consumer_group_848.go b/pkg/kgo/consumer_group_848.go
index 7500d535..f5194c5b 100644
--- a/pkg/kgo/consumer_group_848.go
+++ b/pkg/kgo/consumer_group_848.go
@@ -7,7 +7,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"reflect"
 	"slices"
 	"strings"
 	"time"
@@ -17,12 +16,6 @@ import (
 )
 
 func (g *groupConsumer) should848() bool {
-	if wantBeta := g.cl.ctx.Value("opt_in_kafka_next_gen_balancer_beta"); wantBeta == nil { // !!! TODO REMOVE ONCE BROKER IMPROVES
-		return false
-	}
-	if g.cl.cfg.disableNextGenBalancer {
-		return false
-	}
 	// We pin to v1, introduced in Kafka 4, which fully stabilizes KIP-848.
 	if !g.cl.supportsKIP848v1() {
 		return false
@@ -60,19 +53,8 @@ func (g *groupConsumer) manage848() {
 		g.cooperative.Store(true) // next gen is always cooperative
 	}
 
-	g.mu.Lock()
 	g848 := &g848{g: g, serverAssignor: serverAssignor}
-	g.g848 = g848
-	g.mu.Unlock()
-
-	// v1+ requires the client to generate their own memberID.
-	// On v0, the server provides the ID and we join twice.
-	// We pin to v1+.
-	g.memberGen.store(newStringUUID(), 0) // 0 joins the group
-
 	var consecutiveErrors int
-	var initialFences int
-outer:
 	for {
 		initialHb, err := g848.initialJoin()
 
@@ -103,30 +85,7 @@ outer:
 			}
 		}
 
-		// !!! TODO THIS BLOCK SHOULD NOT BE NECESSARY !!!
-		if errors.Is(err, kerr.FencedMemberEpoch) && initialFences < 1 {
-			lastMember, gen := g.memberGen.load()
-			newMember := newStringUUID()
-			g.memberGen.store(newMember, 0)
-			g.cfg.logger.Log(LogLevelInfo, "received fenced member epoch with epoch 0; clearing member ID to forcefully join as a new member",
-				"group", g.cfg.group,
-				"prior_member_id", lastMember,
-				"prior_generation", gen,
-				"new_member_id", newMember,
-				"new_generation", 0,
-			)
-			after := time.NewTimer(time.Second)
-			select {
-			case <-after.C:
-				initialFences++
-				continue outer
-			case <-g.cl.ctx.Done():
-				after.Stop()
-			}
-		}
-
 		for err == nil {
-			initialFences = 0
 			consecutiveErrors = 0
 			var nowAssigned map[string][]int32
 			// setupAssignedAndHeartbeat
@@ -143,83 +102,27 @@ outer:
 			// logic that handles all edge conditions.
 			_, err = g.setupAssignedAndHeartbeat(initialHb, func() (time.Duration, error) {
 				req := g848.mkreq()
-				dup := *req
-				if reflect.DeepEqual(g848.lastSubscribedTopics, req.SubscribedTopicNames) && reflect.DeepEqual(g848.lastTopics, req.Topics) {
-					req.InstanceID = nil
-					req.RebalanceTimeoutMillis = -1
-					req.ServerAssignor = nil
-					req.SubscribedTopicRegex = nil
-					req.SubscribedTopicNames = nil
-					req.Topics = nil
-				}
 				resp, err := req.RequestWith(g.ctx, g.cl)
 				if err != nil {
 					return g.cfg.heartbeatInterval, err
 				}
-
-				// !!! TODO BEFORE OPTING IN BY DEFAULT !!!
-				// !!! ALSO EVALUATE COMMIT LOGIC       !!!
-				//
-				// See how the tests perform when the beartbeat is duplicated!
-				// I deliberately am not opting into next gen rebalancing until
-				// tests are reliably stable with the following lines uncommented!
-				//
-				// _, _ = resp, err
-				// resp, err = req.RequestWith(g.ctx, g.cl)
-				// if err != nil {
-				// 	return g.cfg.heartbeatInterval, err
-				// }
-				//
-				// !!! TODO DELETE
-
-				err = errCodeMessage(resp.ErrorCode, resp.ErrorMessage)
-				if errors.Is(err, kerr.FencedMemberEpoch) {
-					req = &dup
-					resp, err = req.RequestWith(g.ctx, g.cl)
-					if err != nil {
-						return g.cfg.heartbeatInterval, err
-					}
-				}
-
 				err = errCodeMessage(resp.ErrorCode, resp.ErrorMessage)
 				sleep := time.Duration(resp.HeartbeatIntervalMillis) * time.Millisecond
 				if err != nil {
 					return sleep, err
 				}
-				hbAssigned := g848.handleResp(req, resp)
+				hbAssigned := g848.handleResp(resp)
 				if hbAssigned != nil {
 					err = kerr.RebalanceInProgress
 					nowAssigned = hbAssigned
 				}
 				return sleep, err
 			})
-			switch {
-			case errors.Is(err, kerr.FencedMemberEpoch):
-				member, gen := g.memberGen.load()
-				g.cfg.logger.Log(LogLevelInfo, "consumer group heartbeat saw fenced member epoch, abandoning assignment and rejoining",
-					"group", g.cfg.group,
-					"member_id", member,
-					"generation", gen,
-				)
-				nowAssigned = make(map[string][]int32)
-				g.nowAssigned.store(nil)
-				continue outer
-
-			case errors.Is(err, kerr.UnknownMemberID):
-				g.memberGen.store(newStringUUID(), 0)
-
-			case errors.Is(err, kerr.RebalanceInProgress):
+			if errors.Is(err, kerr.RebalanceInProgress) {
 				err = nil
 			}
-
 			if nowAssigned != nil {
-				member, gen := g.memberGen.load()
-				g.cfg.logger.Log(LogLevelInfo, "consumer group heartbeat detected an updated assignment; exited heartbeat loop to assign & reentering",
-					"group", g.cfg.group,
-					"member_id", member,
-					"generation", gen,
-					"now_assigned", nowAssigned,
-				)
+				g.cfg.logger.Log(LogLevelInfo, "consumer group heartbeat detected an updated assignment; exited heartbeat loop to assign & reentering", "now_assigned", nowAssigned)
 				g.nowAssigned.store(nowAssigned)
 			}
 		}
@@ -281,7 +184,7 @@ type g848 struct {
 	serverAssignor string
 
 	lastSubscribedTopics []string
-	lastTopics           []kmsg.ConsumerGroupHeartbeatRequestTopic
+	lastTopics           map[string][]int32
 }
 
 // v1+ requires the end user to generate their own MemberID, with the
@@ -295,9 +198,10 @@ func newStringUUID() string {
 }
 
 func (g *g848) initialJoin() (time.Duration, error) {
-	g.g.memberGen.storeGeneration(0)
-	g.lastSubscribedTopics = nil
-	g.lastTopics = nil
+	// v1+ requires the client to generate their own memberID.
+	// On v0, the server provides the ID and we join twice.
+	// We pin to v1+.
+	g.g.memberGen.store(newStringUUID(), 0) // 0 joins the group
 	req := g.mkreq()
 	resp, err := req.RequestWith(g.g.ctx, g.g.cl)
 	if err == nil {
@@ -306,15 +210,8 @@ func (g *g848) initialJoin() (time.Duration, error) {
 	if err != nil {
 		return 0, err
 	}
-	nowAssigned := g.handleResp(req, resp)
-	member, gen := g.g.memberGen.load()
-	g.g.cfg.logger.Log(LogLevelInfo, "consumer group initial heartbeat received assignment",
-		"group", g.g.cfg.group,
-		"member_id", member,
-		"generation", gen,
-		"now_assigned", nowAssigned,
-	)
-
+	nowAssigned := g.handleResp(resp)
+	g.g.cfg.logger.Log(LogLevelInfo, "consumer group initial heartbeat received assignment", "now_assigned", nowAssigned)
 	// We always keep nowAssigned: the field is always nil here, so we
 	// either re-store nil or we save an update.
 	if g.g.nowAssigned.read() != nil {
@@ -324,25 +221,19 @@ func (g *g848) initialJoin() (time.Duration, error) {
 	return time.Duration(resp.HeartbeatIntervalMillis) * time.Millisecond, nil
 }
 
-func (g *g848) handleResp(req *kmsg.ConsumerGroupHeartbeatRequest, resp *kmsg.ConsumerGroupHeartbeatResponse) map[string][]int32 {
+func (g *g848) handleResp(resp *kmsg.ConsumerGroupHeartbeatResponse) map[string][]int32 {
 	if resp.MemberID != nil {
 		g.g.memberGen.store(*resp.MemberID, resp.MemberEpoch)
-		g.g.cl.cfg.logger.Log(LogLevelDebug, "storing member and epoch", "member", *resp.MemberID, "epoch", resp.MemberEpoch)
 	} else {
 		g.g.memberGen.storeGeneration(resp.MemberEpoch)
-		g.g.cl.cfg.logger.Log(LogLevelDebug, "storing epoch", "epoch", resp.MemberEpoch)
 	}
 
-	id2t := g.g.cl.id2tMap()
-	newAssigned := make(map[string][]int32)
-
-	g.lastSubscribedTopics = req.SubscribedTopicNames
-	g.lastTopics = req.Topics
-
 	if resp.Assignment == nil {
 		return nil
 	}
 
+	id2t := g.g.cl.id2tMap()
+	newAssigned := make(map[string][]int32)
 	for _, t := range resp.Assignment.Topics {
 		name := id2t[t.TopicID]
 		if name == "" {
@@ -356,7 +247,6 @@ func (g *g848) handleResp(req *kmsg.ConsumerGroupHeartbeatRequest, resp *kmsg.Co
 		slices.Sort(t.Partitions)
 		newAssigned[name] = t.Partitions
 	}
-
 	current := g.g.nowAssigned.read()
 	if !mapi32sDeepEq(current, newAssigned) {
 		return newAssigned
@@ -371,8 +261,7 @@ func (g *g848) mkreq() *kmsg.ConsumerGroupHeartbeatRequest {
 
 	// Most fields in the request can be null if the field is equal to the
 	// last time we sent the request. The first time we write, we include
-	// all information. We always return all information here; the caller
-	// of mkreq may strip fields as needed.
+	// all information.
 	//
 	// Our initial set of subscribed topics is specified in our config.
 	// For non-regex consuming, topics are directly created into g.tps.
@@ -383,39 +272,76 @@ func (g *g848) mkreq() *kmsg.ConsumerGroupHeartbeatRequest {
 	// For regex topics, they cannot add or remove after client creation.
 	// We just use the initial config field.
 
-	req.InstanceID = g.g.cfg.instanceID
-	if g.g.cfg.rack != "" {
-		req.RackID = &g.g.cfg.rack
-	}
-	req.RebalanceTimeoutMillis = int32(g.g.cfg.rebalanceTimeout.Milliseconds())
-	req.ServerAssignor = &g.serverAssignor
+	switch req.MemberEpoch {
+	case 0:
+		req.InstanceID = g.g.cfg.instanceID
+		if g.g.cfg.rack != "" {
+			req.RackID = &g.g.cfg.rack
+		}
+		req.RebalanceTimeoutMillis = int32(g.g.cfg.rebalanceTimeout.Milliseconds())
+		req.ServerAssignor = &g.serverAssignor
+
+		tps := g.g.tps.load()
+		if g.g.cl.cfg.regex {
+			topics := g.g.cl.cfg.topics
+			patterns := make([]string, 0, len(topics))
+			for topic := range topics {
+				patterns = append(patterns, "(?:"+topic+")")
+			}
+			pattern := strings.Join(patterns, "|")
+			req.SubscribedTopicRegex = &pattern
+		} else {
+			// SubscribedTopics must always exist when epoch == 0.
+			// We specifically 'make' the slice to ensure it is non-nil.
+			subscribedTopics := make([]string, 0, len(tps))
+			for t := range tps {
+				subscribedTopics = append(subscribedTopics, t)
+			}
+			g.lastSubscribedTopics = subscribedTopics
+			req.SubscribedTopicNames = subscribedTopics
+		}
 
-	tps := g.g.tps.load()
-	if g.g.cl.cfg.regex {
-		topics := g.g.cl.cfg.topics
-		patterns := make([]string, 0, len(topics))
-		for topic := range topics {
-			patterns = append(patterns, "(?:"+topic+")")
+		// Topics must be empty when we do the initial join.
+		// We sanity-check that we are not assigned anything.
+		if len(g.g.nowAssigned.read()) > 0 {
+			panic("issuing ConsumerGroupHeartbeat while we are currently assigned something, bug, please open an issue!")
 		}
-		pattern := strings.Join(patterns, "|")
-		req.SubscribedTopicRegex = &pattern
-	} else {
-		// SubscribedTopics must always exist when epoch == 0.
-		// We specifically 'make' the slice to ensure it is non-nil.
-		subscribedTopics := make([]string, 0, len(tps))
-		for t := range tps {
-			subscribedTopics = append(subscribedTopics, t)
+		g.lastTopics = noConsumerGroupHeartbeatReqTopics
+		req.Topics = []kmsg.ConsumerGroupHeartbeatRequestTopic{} // must always exist and be empty when joining
+
+	default:
+		tps := g.g.tps.load()
+
+		// SubscribedTopicRegex never changes, so we never need to
+		// re-include it.
+		if !g.g.cl.cfg.regex {
+			subscribedTopics := make([]string, 0, len(tps))
+			for t := range tps {
+				subscribedTopics = append(subscribedTopics, t)
+			}
+			if !slicesDeepEq(g.lastSubscribedTopics, subscribedTopics) {
+				req.SubscribedTopicNames = subscribedTopics
+			}
+			g.lastSubscribedTopics = subscribedTopics
 		}
-		req.SubscribedTopicNames = subscribedTopics
-	}
-	nowAssigned := g.g.nowAssigned.clone()                   // always returns non-nil
-	req.Topics = []kmsg.ConsumerGroupHeartbeatRequestTopic{} // ALWAYS initialize: len 0 is significantly different than nil (nil means same as last time)
-	for t, ps := range nowAssigned {
-		rt := kmsg.NewConsumerGroupHeartbeatRequestTopic()
-		rt.Partitions = slices.Clone(ps)
-		rt.TopicID = tps[t].load().id
-		req.Topics = append(req.Topics, rt)
+
+		nowAssigned := g.g.nowAssigned.clone() // always returns non-nil
+		if !mapi32sDeepEq(g.lastTopics, nowAssigned) {
+			req.Topics = []kmsg.ConsumerGroupHeartbeatRequestTopic{} // ALWAYS initialize: len 0 is significantly different than nil (nil means same as last time)
+			tps := g.g.tps.load()
+			for t, ps := range nowAssigned {
+				rt := kmsg.NewConsumerGroupHeartbeatRequestTopic()
+				rt.Partitions = slices.Clone(ps)
+				rt.TopicID = tps[t].load().id
+				req.Topics = append(req.Topics, rt)
+			}
+		}
+		g.lastTopics = nowAssigned
 	}
 
+	// TODO regex v1+
+
 	return req
 }
+
+var noConsumerGroupHeartbeatReqTopics = make(map[string][]int32)
diff --git a/pkg/kgo/errors.go b/pkg/kgo/errors.go
index bbc670e6..35322a03 100644
--- a/pkg/kgo/errors.go
+++ b/pkg/kgo/errors.go
@@ -11,18 +11,6 @@ import (
 	"github.com/twmb/franz-go/pkg/kerr"
 )
 
-// IsRetryableBrokerErr returns whether the client considers an error from a
-// broker retrayble. This returns true specifically if the client thinks it can
-// retry whatever it was just trying to do with a broker. It returns false in
-// all other cases.
-//
-// This can used external to the library to help filter errors if use kgo
-// hooks: errors may be sent to hooks before the client retries whatever it was
-// just attempting.
-func IsRetryableBrokerErr(err error) bool {
-	return isRetryableBrokerErr(err)
-}
-
 func isRetryableBrokerErr(err error) bool {
 	// The error could be nil if we are evaluating multiple errors at once,
 	// and only one is non-nil. The intent of this function is to evaluate
@@ -277,20 +265,15 @@ type ErrDataLoss struct {
 	// ConsumedTo is what the client had consumed to for this partition before
 	// data loss was detected.
 	ConsumedTo int64
-	// ConsumedToEpoch is the epoch for the offset the client was currently
-	// consuming.
-	ConsumedToEpoch int32
 	// ResetTo is what the client reset the partition to; everything from
 	// ResetTo to ConsumedTo was lost.
 	ResetTo int64
-	// ResetToEpoch is the epoch the client was reset to.
-	ResetToEpoch int32
 }
 
 func (e *ErrDataLoss) Error() string {
 	return fmt.Sprintf("topic %s partition %d lost records;"+
-		" the client consumed to offset %d epoch %d but was reset to offset %d epoch %d",
-		e.Topic, e.Partition, e.ConsumedTo, e.ConsumedToEpoch, e.ResetTo, e.ResetToEpoch)
+		" the client consumed to offset %d but was reset to offset %d",
+		e.Topic, e.Partition, e.ConsumedTo, e.ResetTo)
 }
 
 type errUnknownController struct {
@@ -357,7 +340,7 @@ func isDecompressErr(err error) bool {
 func errCodeMessage(code int16, errMessage *string) error {
 	if err := kerr.ErrorForCode(code); err != nil {
 		if errMessage != nil {
-			return fmt.Errorf("%w: %s", err, *errMessage)
+			return fmt.Errorf("%s: %w", *errMessage, err)
 		}
 		return err
 	}
diff --git a/pkg/kgo/hooks.go b/pkg/kgo/hooks.go
index 8edd539c..253c143d 100644
--- a/pkg/kgo/hooks.go
+++ b/pkg/kgo/hooks.go
@@ -290,7 +290,7 @@ type FetchBatchMetrics struct {
 	CompressionType uint8
 }
 
-// HookFetchBatchRead is called whenever a batch is read within the client.
+// HookFetchBatchRead is called whenever a batch if read within the client.
 //
 // Note that this hook is called when processing, but a batch may be internally
 // discarded after processing in some uncommon specific circumstances.
diff --git a/pkg/kgo/metrics_714.go b/pkg/kgo/metrics_714.go
deleted file mode 100644
index a238f8c8..00000000
--- a/pkg/kgo/metrics_714.go
+++ /dev/null
@@ -1,990 +0,0 @@
-package kgo
-
-import (
-	"bytes"
-	"context"
-	"encoding/binary"
-	"errors"
-	"fmt"
-	"math"
-	"math/rand"
-	"strings"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/twmb/franz-go/pkg/kerr"
-	"github.com/twmb/franz-go/pkg/kmsg"
-)
-
-func (cl *Client) pushMetrics() {
-	defer cl.metrics.ctxCancel()
-
-	if cl.cfg.disableClientMetrics {
-		return
-	}
-
-	m := &cl.metrics
-
-	select {
-	case <-cl.ctx.Done():
-		return
-	case <-m.firstObserve:
-	}
-
-	if !cl.supportsClientMetrics() {
-		m.mu.Lock()
-		m.unsupported.Store(true)
-		m.pReqLatency = nil
-		m.cReqLatency = nil
-		m.mu.Unlock()
-		return
-	}
-
-	var clientInstanceID [16]byte
-	var terminating bool
-	for !terminating {
-		greq := kmsg.NewGetTelemetrySubscriptionsRequest()
-		greq.ClientInstanceID = clientInstanceID
-
-		gresp, err := greq.RequestWith(m.ctx, cl)
-		if err == nil {
-			err = kerr.ErrorForCode(gresp.ErrorCode)
-		}
-		if err != nil {
-			cl.cfg.logger.Log(LogLevelInfo, "unable to get telemetry subscriptions, retrying in 30s", "err", err)
-			after := time.NewTimer(30 * time.Second)
-			select {
-			case <-cl.ctx.Done():
-				after.Stop()
-				return
-			case <-after.C:
-			}
-			continue
-		}
-
-		clientInstanceID = gresp.ClientInstanceID
-
-		// If there are no requested metrics, we wait the push interval
-		// and re-get.
-		if len(gresp.RequestedMetrics) == 0 {
-			wait := time.Duration(gresp.PushIntervalMillis) * time.Millisecond
-			cl.cfg.logger.Log(LogLevelInfo, "no metrics requested, sleeping and asking again later", "sleep", wait)
-			after := time.NewTimer(wait)
-			select {
-			case <-cl.ctx.Done():
-				terminating = true
-			case <-after.C:
-			}
-			continue
-		}
-		cl.cfg.logger.Log(LogLevelInfo, "received client metrics subscription, beginning periodic send loop",
-			"client_instance_id", fmt.Sprintf("%x", gresp.ClientInstanceID),
-			"subscription_id", gresp.SubscriptionID,
-			"accepteded_compression_types", gresp.AcceptedCompressionTypes,
-			"telemetry_max_bytes", gresp.TelemetryMaxBytes,
-			"push_interval", time.Duration(gresp.PushIntervalMillis)*time.Millisecond,
-			"requested_metrics", gresp.RequestedMetrics,
-		)
-
-		var codecs []CompressionCodec
-		for _, accepted := range gresp.AcceptedCompressionTypes {
-			switch CompressionCodecType(accepted) {
-			case CodecNone:
-			case CodecGzip:
-				codecs = append(codecs, GzipCompression())
-			case CodecSnappy:
-				codecs = append(codecs, SnappyCompression())
-			case CodecLz4:
-				codecs = append(codecs, Lz4Compression())
-			case CodecZstd:
-				codecs = append(codecs, ZstdCompression())
-			}
-		}
-		compressor, _ := DefaultCompressor(codecs...)
-		allowedNames := buildNameFilter(gresp.RequestedMetrics)
-
-		// We want to pin pushing metrics to a single broker until that
-		// broker goes away. To do this, we first issue RequestSharded,
-		// figure out the broker that responded, and then use that
-		// broker until we receive errUnknownBroker.
-		var br *Broker
-
-	push:
-		for i := 0; !terminating; i++ {
-			wait := time.Duration(gresp.PushIntervalMillis) * time.Millisecond
-			if wait < time.Second {
-				wait = time.Second
-			}
-			if i == 0 { // for the first request, jitter 0.5 <= wait <= 1.5
-				cl.rng(func(r *rand.Rand) {
-					wait = time.Duration(float64(wait) * (0.5 + r.Float64()))
-				})
-			}
-
-			// Wait until our push interval; if the client is quitting,
-			// we immediately send a push with Terminating=true.
-			after := time.NewTimer(wait)
-			var terminating bool
-			select {
-			case <-cl.ctx.Done():
-				terminating = true
-			case <-after.C:
-			}
-
-			// Create our request.
-			preq := kmsg.NewPushTelemetryRequest()
-			preq.ClientInstanceID = clientInstanceID
-			preq.SubscriptionID = gresp.SubscriptionID
-			preq.Terminating = terminating
-			serialized, compression, nmetrics := m.appendTo(nil, gresp.DeltaTemporality, gresp.TelemetryMaxBytes, allowedNames, compressor)
-			if len(serialized) > int(gresp.TelemetryMaxBytes) {
-				cl.cfg.logger.Log(LogLevelWarn, "serialized metrics are larger than the broker provided max limit, sending anyway and hoping for the best",
-					"max_bytes", gresp.TelemetryMaxBytes,
-					"serialized_bytes", len(serialized),
-				)
-			}
-			preq.CompressionType = compression
-			preq.Metrics = serialized
-
-			cl.cfg.logger.Log(LogLevelDebug, "sending client metrics to broker", "num_metrics", nmetrics)
-
-			// Send our request, pinning to the broker we get a response
-			// from or using the pinned broker if possible.
-			var resp kmsg.Response
-			var err error
-		doreq:
-			if br == nil {
-				shard := cl.RequestSharded(m.ctx, &preq)[0]
-				resp, err = shard.Resp, shard.Err
-				if err == nil {
-					br = cl.Broker(int(shard.Meta.NodeID))
-				}
-			} else {
-				resp, err = br.RetriableRequest(m.ctx, &preq)
-				if errors.Is(err, errUnknownBroker) {
-					br = nil
-					goto doreq
-				}
-			}
-			if err != nil {
-				cl.cfg.logger.Log(LogLevelWarn, "unable to send client metrics, resetting subscription", "err", err)
-				break
-			}
-
-			// Not much to do on the response besides a bunch of
-			// error handling.
-			presp := resp.(*kmsg.PushTelemetryResponse)
-			switch err := kerr.ErrorForCode(presp.ErrorCode); err {
-			case nil:
-			case kerr.InvalidRequest:
-				cl.cfg.logger.Log(LogLevelError, "client metrics was sent at the wrong time (after sending terminating request already?), exiting metrics loop", "err", err)
-				return
-			case kerr.InvalidRecord:
-				cl.cfg.logger.Log(LogLevelError, "broker could not understand our client metrics serialization, this is perhaps a franz-go bug", "err", err)
-				return
-			case kerr.TelemetryTooLarge:
-				cl.cfg.logger.Log(LogLevelWarn, "client metrics payload was too large (check your broker max telemetry bytes)", "err", err)
-				// Do nothing; continue aggregating
-			case kerr.UnknownSubscriptionID:
-				cl.cfg.logger.Log(LogLevelInfo, "client metrics has an outdated subscription, re-getting our subscription information", "err", err)
-				break push
-			case kerr.UnsupportedCompressionType:
-				cl.cfg.logger.Log(LogLevelInfo, "client metrics compression is not supported by the broker even though we only used previously supported compressors, re-getting our subscription information", "err", err)
-			default:
-				if !kerr.IsRetriable(err) {
-					cl.cfg.logger.Log(LogLevelError, "client metrics received an unknown error we do not know how to handle, exiting metrics loop", "err", err)
-					return
-				}
-				cl.cfg.logger.Log(LogLevelWarn, "client metrics received an unknown error that is retryably, continuing to next push cycle", "err", err)
-			}
-		}
-	}
-}
-
-func buildNameFilter(requested []string) func(string) bool {
-	if len(requested) == 0 {
-		return func(string) bool { return false }
-	}
-	if len(requested) == 1 && requested[0] == "" {
-		return func(string) bool { return true }
-	}
-	return func(name string) bool {
-		for _, pfx := range requested {
-			if strings.HasPrefix(name, pfx) {
-				return true
-			}
-		}
-		return false
-	}
-}
-
-const (
-	// MetricTypeSum is a sum type metric. Every time the metric is
-	// collected, the number you are returning is cumulative from the time
-	// the client / your application was initialized.
-	MetricTypeSum = 1 + iota
-
-	// MetricTypeGauge is a gauge type metric. It is a recording of a value
-	// at a point in time.
-	MetricTypeGauge
-)
-
-type (
-	// MetricType is the type of metric you are providing: Type is the type
-	// of metric this is: either a gauge or a sum type.
-	MetricType uint8
-
-	// Metric is a user-defined client side metric so that you can send
-	// user-defined client metrics to the broker and give your cluster
-	// operator insight into your client.
-	//
-	// This type exists to support KIP-1076, which is an extension to
-	// KIP-714. Read either of those for more detail.
-	Metric struct {
-		// Name is a user provided metric name.
-		//
-		// KIP-714 prescribes how to name metrics: lowercase with dots,
-		// no dashes, and interoperable with the OpenTelemetry
-		// ecosystem. It is recommended follow the KIP-714 guidance and
-		// to namespace your metrics ("my.specific.metric.1;
-		// my.specific.metric.2). This client does not attempt to
-		// further santize your user provided name.
-		Name string
-
-		// Type is the type of metric this is: either a gauge or a sum
-		// type.
-		//
-		// Note that for sum types, the client internally caches all
-		// sum types by name for one extra collection period so that
-		// the client can calculate "delta" metrics if the broker
-		// requests them. You must provide the sum type metric on every
-		// collection; skipping a cycle means the client will be unable
-		// to calculate the delta once you provide it again in the
-		// future.
-		Type MetricType
-
-		// ValueInt is the value to record. Only one of ValueInt or
-		// ValueFloat should be non-zero; if both are non-zero or if
-		// both are zero, the metric is ignored.
-		//
-		// Sum metrics only support ValueInt, and the number should
-		// never go down. If the value goes down or ValueFloat is
-		// used, the metric is skipped.
-		ValueInt int64
-
-		// ValueFloat is the value to record. Only one of ValueInt or
-		// ValueFloat should be non-zero; if both are non-zero or if
-		// both are zero, the metric is ignored.
-		//
-		// Sum metrics only support ValueInt, and the number should
-		// never go down. If the value goes down or ValueFloat is
-		// used, the metric is skipped.
-		ValueFloat float64
-
-		// Attrs are optional attributes to add to this metric, such as
-		// a node ID. The supported `any` types are strings, booleans,
-		// numbers, and byte slices. All other attributes are silently
-		// skipped. Attributes should be lowercase with underscores
-		// (see KIP-714).
-		Attrs map[string]any
-	}
-
-	// metricRate is a count per second and a total.
-	metricRate struct {
-		count   atomic.Int64 // Sum of events this period; rate == float64(count/time) at rollup
-		tot     atomic.Int64 // Total events over all time.
-		lastTot int64        // Updated when encoding; the last value for tot in case broker requests DELTA.
-	}
-
-	// metricTime reports average latency, max latency, and total latency.
-	// The unit is in milliseconds.
-	metricTime struct {
-		sum atomic.Int64 // With separate aggDur field, avg = sum/aggDur at rollup.
-		max atomic.Int64 // Max latency seen during this window.
-	}
-
-	// We skip:
-	// * producer.record.queue.time.{avg,max}     : medium signal; requires more wiring in the producer
-	// * consumer.poll.idle.ratio.avg             : less relevant in this client
-	// * consumer.coordinator.rebalance.latency   : underspecified: should this just track leader rebalance time? or from when we notice rebalance in progress to the next ok heartbeat? or..?
-	// * consumer.fetch.manager.fetch.latency     : seems to duplicate consumer.node.request.latency ??
-	// * consumer.coordinator.assigned.partitions : honestly just tedious to work in given incremental changes from cooperative rebalancing; open to being convinced to add this
-	metrics struct {
-		cl *Client
-
-		closedFirstObserve atomic.Bool
-
-		// mu is grabbed when accessing the map fields.
-		mu          sync.Mutex
-		unsupported atomic.Bool // set to true if the broker does not support client metrics; guards nil-ing the maps
-
-		pConnCreation metricRate
-		pReqLatency   map[int32]*metricTime
-		pThrottle     metricTime
-
-		cConnCreation  metricRate
-		cReqLatency    map[int32]*metricTime
-		cCommitLatency metricTime
-
-		initNano     int64
-		lastPushNano int64
-
-		userSumLast map[string]int64
-
-		firstObserve chan struct{}
-
-		ctx       context.Context
-		ctxCancel func()
-	}
-)
-
-func (m *metrics) init(cl *Client) {
-	m.cl = cl
-	m.initNano = time.Now().UnixNano()
-	m.firstObserve = make(chan struct{})
-	m.ctx, m.ctxCancel = context.WithCancel(context.Background()) // for graceful shutdown
-}
-
-func safeDiv[T ~int64 | ~float64](num, denom T) T {
-	if denom == 0 {
-		return 0
-	}
-	return num / denom
-}
-
-// Internally we use the metrics.observeRate function so that
-// triggerFirstObserve is always called.
-func (t *metricRate) observe() {
-	t.count.Add(1)
-	t.tot.Add(1)
-}
-
-func (t *metricRate) rollNums() (rate float64, tot, lastTot int64) {
-	count := t.count.Swap(0)
-	lastTot = t.lastTot
-	tot = t.tot.Load()
-	t.lastTot = tot
-
-	rate = safeDiv(float64(count), float64(tot-lastTot))
-	return rate, tot, lastTot
-}
-
-// Internally we use the metrics.observeTime function so that
-// triggerFirstObserve is always called.
-func (t *metricTime) observe(millis int64) {
-	t.sum.Add(millis)
-	for {
-		max := t.max.Load()
-		if millis < max {
-			return
-		}
-		if t.max.CompareAndSwap(max, millis) {
-			return
-		}
-	}
-}
-
-func (t *metricTime) rollNums(aggDur time.Duration) (avg float64, max int64) {
-	sum := t.sum.Swap(0)
-	max = t.max.Swap(0)
-	avg = safeDiv(float64(sum), float64(aggDur.Milliseconds()))
-	return avg, max
-}
-
-func (m *metrics) triggerFirstObserve() {
-	if !m.closedFirstObserve.Swap(true) {
-		close(m.firstObserve)
-	}
-}
-
-func (m *metrics) observeRate(field *metricRate) {
-	m.triggerFirstObserve()
-	field.observe()
-}
-
-func (m *metrics) observeTime(field *metricTime, millis int64) {
-	m.triggerFirstObserve()
-	field.observe(millis)
-}
-
-func (m *metrics) observeNodeTime(node int32, field *map[int32]*metricTime, millis int64) {
-	m.triggerFirstObserve()
-	if m.unsupported.Load() {
-		return
-	}
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.unsupported.Load() { // one more check inside the lock to avoid a race where the maps are set nil after storing true
-		return
-	}
-	if *field == nil {
-		*field = make(map[int32]*metricTime)
-	}
-	t := (*field)[node]
-	if t == nil {
-		t = new(metricTime)
-		(*field)[node] = t
-	}
-	t.observe(millis)
-}
-
-func (m *metrics) appendTo(b []byte, useDeltaSums bool, maxBytes int32, allowedNames func(string) bool, compressor Compressor) ([]byte, int8, int) {
-	////////////
-	// VARIABLE INITIALIZATION
-	////////////
-	var (
-		metricsData    otelMetricsData
-		resourceMetric = &metricsData.resourceMetric
-		resource       = &resourceMetric.resource
-		scopeMetric    = &resourceMetric.scopeMetric
-		scope          = &scopeMetric.scope
-		metrics        = &scopeMetric.metrics
-		labels         = make(map[string]any)
-	)
-
-	if m.cl.cfg.rack != "" {
-		labels["client_rack"] = m.cl.cfg.rack
-	}
-	if m.cl.cfg.group != "" {
-		labels["group_id"] = m.cl.cfg.group
-	}
-	if m.cl.cfg.instanceID != nil {
-		labels["group_instance_id"] = *m.cl.cfg.instanceID
-	}
-	if memberID, _ := m.cl.GroupMetadata(); memberID != "" {
-		labels["group_member_id"] = memberID
-	}
-	if m.cl.cfg.txnID != nil {
-		labels["transactional_id"] = *m.cl.cfg.txnID
-	}
-	if len(labels) > 0 {
-		resource.attributes = labels
-	}
-	scope.name = "kgo"
-	scope.version = softwareVersion()
-
-	now := time.Now()
-	nowNano := now.UnixNano()
-	lastPush := m.lastPushNano
-	aggDur := now.Sub(time.Unix(0, lastPush))
-	if lastPush == 0 {
-		aggDur = now.Sub(time.Unix(0, m.initNano))
-	}
-	m.lastPushNano = nowNano
-
-	////////////
-	// FUNCTIONS THAT APPEND TO `metrics`.
-	////////////
-
-	appendGauge := func(name string, vi64 int64, vf64 float64, attrs map[string]any) {
-		if vi64 == 0 && vf64 == 0 || vi64 != 0 && vf64 != 0 {
-			return
-		}
-		if !allowedNames(name) {
-			return
-		}
-		*metrics = append(*metrics, otelMetric{
-			name: name,
-			gauge: otelGauge{
-				otelNumDataPoint{
-					attributes: attrs,
-					vInt:       vi64,
-					vDouble:    vf64,
-					startNano:  lastPush,
-					timeNano:   nowNano,
-				},
-			},
-		})
-	}
-	appendSum := func(name string, tot, lastTot int64, attrs map[string]any) {
-		if tot-lastTot == 0 {
-			return
-		}
-		if !allowedNames(name) {
-			return
-		}
-		if useDeltaSums {
-			*metrics = append(*metrics, otelMetric{
-				name: name,
-				sum: otelSum{
-					dataPoint: otelNumDataPoint{
-						attributes: attrs,
-						vInt:       tot - lastTot,
-						startNano:  lastPush,
-						timeNano:   nowNano,
-					},
-					aggregationTemporality: otelTempDelta,
-				},
-			})
-		} else {
-			*metrics = append(*metrics, otelMetric{
-				name: name,
-				sum: otelSum{
-					dataPoint: otelNumDataPoint{
-						attributes: attrs,
-						vInt:       tot,
-						startNano:  m.initNano,
-						timeNano:   nowNano,
-					},
-					aggregationTemporality: otelTempCumulative,
-				},
-			})
-		}
-	}
-
-	////////////
-	// COLLECTING ALL METRICS TO SEND
-	////////////
-
-	m.mu.Lock()
-	for _, s := range []struct {
-		name string
-		v    any
-	}{
-		{"org.apache.kafka.producer.connection.creation", &m.pConnCreation},
-		{"org.apache.kafka.producer.node.request.latency", &m.pReqLatency},
-		{"org.apache.kafka.producer.produce.throttle.time", &m.pThrottle},
-		{"org.apache.kafka.consumer.connection.creation", &m.cConnCreation},
-		{"org.apache.kafka.consumer.node.request.latency", &m.cReqLatency},
-		{"org.apache.kafka.consumer.coordinator.commit.latency", &m.cCommitLatency},
-	} {
-		switch t := s.v.(type) {
-		case *metricRate:
-			rate, tot, lastTot := t.rollNums()
-			appendGauge(s.name+".rate", 0, rate, nil)
-			appendSum(s.name+".total", tot, lastTot, nil)
-
-		case *metricTime:
-			avg, max := t.rollNums(aggDur)
-			appendGauge(s.name+".avg", 0, avg, nil)
-			appendGauge(s.name+".max", max, 0, nil)
-
-		case *map[int32]*metricTime:
-			for broker, m := range *t {
-				avg, max := m.rollNums(aggDur)
-				attrs := map[string]any{"node_id": broker}
-				appendGauge(s.name+".avg", 0, avg, attrs)
-				appendGauge(s.name+".max", max, 0, attrs)
-				// By-node metrics do not write the forever-total-latency.
-				// We only have per-push increments, so we can safely delete
-				// the node every round (i.e., clean up if a broker goes away).
-				if avg == 0 && max == 0 {
-					delete(*t, broker)
-				}
-			}
-
-		default:
-			m.mu.Unlock()
-			panic("unsupported type")
-		}
-	}
-	m.mu.Unlock()
-
-	userAt := len(*metrics)
-	if m.cl.cfg.userMetrics != nil {
-		var userSkipped []string
-		last := m.userSumLast
-		if last == nil {
-			last = make(map[string]int64)
-		}
-		m.userSumLast = make(map[string]int64)
-		for um := range m.cl.cfg.userMetrics() {
-			if um.ValueInt != 0 && um.ValueFloat != 0 || um.ValueInt == 0 && um.ValueFloat == 0 {
-				userSkipped = append(userSkipped, um.Name)
-				continue
-			}
-
-			switch um.Type {
-			case MetricTypeSum:
-				lastTot := last[um.Name]
-				if um.ValueFloat != 0 || um.ValueInt < 0 || lastTot > um.ValueInt {
-					userSkipped = append(userSkipped, um.Name)
-					continue
-				}
-				appendSum(um.Name, um.ValueInt, lastTot, um.Attrs)
-				m.userSumLast[um.Name] = um.ValueInt
-
-			case MetricTypeGauge:
-				appendGauge(um.Name, um.ValueInt, um.ValueFloat, um.Attrs)
-			}
-		}
-		if len(userSkipped) > 0 {
-			m.cl.cfg.logger.Log(LogLevelWarn, "skipped serialization of some user provided metrics", "skipped", userSkipped)
-		}
-	}
-
-	////////////
-	// SERIALIZING - finally append metricsData to b
-	////////////
-
-	serialized, codec := metricsData.appendTo(b[:0]), CodecNone
-
-	if compressor != nil {
-		serialized, codec = compressor.Compress(new(bytes.Buffer), serialized)
-	}
-
-	if len(serialized) > int(maxBytes) && userAt < len(*metrics) {
-		m.cl.cfg.logger.Log(LogLevelWarn, "adding user metrics results in a too-large payload; dropping user metrics and serializing only standard client metrics",
-			"max_bytes", maxBytes,
-			"serialized_bytes_with_user", len(serialized),
-		)
-		*metrics = (*metrics)[:userAt]
-		serialized, codec = metricsData.appendTo(b[:0]), CodecNone
-		if compressor != nil {
-			serialized, codec = compressor.Compress(new(bytes.Buffer), serialized)
-		}
-	}
-	return serialized, int8(codec), len(*metrics)
-}
-
-// The types below are encoded in protobuf format; canonical
-// definitions can be found at:
-// https://github.com/open-telemetry/opentelemetry-proto/blob/f24da8deeb50118271c9435972791ef05ec003b1/opentelemetry/proto/metrics/v1/metrics.proto
-type (
-	otelMetricsData struct {
-		// resourceMetrics is repeated, but we always use one element.
-		resourceMetric otelResourceMetric // = 1
-	}
-
-	otelResourceMetric struct {
-		// We do not encode resource if resource.attributes is empty;
-		// eliding resource just means it is not known.
-		resource otelResource // = 1
-
-		// scopeMetric is repeated, but we always use one element.
-		scopeMetric otelScopeMetric // = 2
-	}
-
-	// From a separate file:
-	// https://github.com/open-telemetry/opentelemetry-proto/blob/f24da8deeb50118271c9435972791ef05ec003b1/opentelemetry/proto/resource/v1/resource.proto
-	// We only use attributes, and only if any exist.
-	otelResource struct {
-		attributes map[string]any // = 1
-	}
-
-	otelScopeMetric struct {
-		scope   otelInstrumentationScope // = 1
-		metrics []otelMetric             // = 2
-	}
-
-	// From a separate file:
-	// https://github.com/open-telemetry/opentelemetry-proto/blob/f24da8deeb50118271c9435972791ef05ec003b1/opentelemetry/proto/common/v1/common.proto
-	// We use "kgo" and our released version (via the `softwareVersion()` func).
-	otelInstrumentationScope struct {
-		name    string // = 1
-		version string // = 2
-	}
-
-	otelMetric struct {
-		name string // = 1
-
-		// The metric data can be oneOf the following two:
-		gauge otelGauge // = 5
-		sum   otelSum   // = 7
-		// We use the non-empty struct; one must be non-empty.
-	}
-
-	otelGauge struct {
-		// The .proto defines this as `repeated`, but we always use
-		// only one data point.
-		dataPoint otelNumDataPoint // = 1
-	}
-
-	otelSum struct {
-		// Same as otelGauge, the .proto defines this as `repeated`
-		// but we only use one data point.
-		dataPoint otelNumDataPoint // = 1
-
-		// aggregationTemporality is an enum;
-		// 0 is unspecified
-		// 1 is delta
-		// 2 is cumulative
-		aggregationTemporality uint8 // = 2
-
-		// We always set isMonotonic to true.
-		isMonotonic bool // = 3
-	}
-
-	otelNumDataPoint struct {
-		// Encoded as key/value attributes; we currently only use int32
-		// broker IDs. This will need to change if other types are
-		// needed.
-		attributes map[string]any // = 7
-
-		startNano int64 // = 2
-		timeNano  int64 // = 3
-
-		// Only one of vDouble or vInt is non-zero, only
-		// one is used (this is oneof).
-		vDouble float64 // = 4
-		vInt    int64   // = 6
-	}
-)
-
-const (
-	otelTempDelta      = 1
-	otelTempCumulative = 2
-)
-
-////////////
-// SERIALIZATION
-////////////
-
-const (
-	protoTypeVarint = 0
-	protoType64bit  = 1
-	protoTypeLength = 2
-	protoType32bit  = 5
-)
-
-// appendProtoTag adds a Protocol Buffer tag (field number + proto type)
-func appendProtoTag(b []byte, fieldNumber, protoType int) []byte {
-	return binary.AppendUvarint(b, uint64((fieldNumber<<3)|protoType))
-}
-
-// appendProtoString appends a string as length-prefixed bytes
-func appendProtoString(b []byte, s string) []byte {
-	b = binary.AppendUvarint(b, uint64(len(s)))
-	return append(b, s...)
-}
-
-func (d *otelMetricsData) appendTo(b []byte) []byte {
-	// Field 1: resourceMetric (message)
-	b = appendProtoTag(b, 1, protoTypeLength)
-	resourceBytes := d.resourceMetric.appendTo(nil)
-	b = binary.AppendUvarint(b, uint64(len(resourceBytes)))
-	b = append(b, resourceBytes...)
-
-	return b
-}
-
-func (m *otelResourceMetric) appendTo(b []byte) []byte {
-	// Field 1: resource (message) - only if attributes are present
-	if len(m.resource.attributes) > 0 {
-		b = appendProtoTag(b, 1, protoTypeLength)
-		resourceBytes := m.resource.appendTo(nil)
-		b = binary.AppendUvarint(b, uint64(len(resourceBytes)))
-		b = append(b, resourceBytes...)
-	}
-
-	// Field 2: scopeMetric (message)
-	b = appendProtoTag(b, 2, protoTypeLength)
-	scopeBytes := m.scopeMetric.appendTo(nil)
-	b = binary.AppendUvarint(b, uint64(len(scopeBytes)))
-	b = append(b, scopeBytes...)
-
-	return b
-}
-
-func (r *otelResource) appendTo(b []byte) []byte {
-	// Field 1: attributes (repeated KeyValue)
-	return appendOtelAttributesTo(b, 1, r.attributes)
-}
-
-func (s *otelScopeMetric) appendTo(b []byte) []byte {
-	// Field 1: scope (message)
-	b = appendProtoTag(b, 1, protoTypeLength)
-	scopeBytes := s.scope.appendTo(nil)
-	b = binary.AppendUvarint(b, uint64(len(scopeBytes)))
-	b = append(b, scopeBytes...)
-
-	// Field 2: metrics (repeated message)
-	for _, m := range s.metrics {
-		b = appendProtoTag(b, 2, protoTypeLength)
-		metricBytes := m.appendTo(nil)
-		b = binary.AppendUvarint(b, uint64(len(metricBytes)))
-		b = append(b, metricBytes...)
-	}
-
-	return b
-}
-
-func (s *otelInstrumentationScope) appendTo(b []byte) []byte {
-	// Field 1: name (string)
-	if s.name != "" {
-		b = appendProtoTag(b, 1, protoTypeLength)
-		b = appendProtoString(b, s.name)
-	}
-	// Field 2: version (string)
-	if s.version != "" {
-		b = appendProtoTag(b, 2, protoTypeLength)
-		b = appendProtoString(b, s.version)
-	}
-	return b
-}
-
-func (m *otelMetric) appendTo(b []byte) []byte {
-	// Field 1: name (string)
-	if m.name != "" {
-		b = appendProtoTag(b, 1, protoTypeLength)
-		b = appendProtoString(b, m.name)
-	}
-
-	// Field 5: gauge (message) - if used
-	if m.gauge.dataPoint.timeNano != 0 {
-		b = appendProtoTag(b, 5, protoTypeLength)
-		gaugeBytes := m.gauge.appendTo(nil)
-		b = binary.AppendUvarint(b, uint64(len(gaugeBytes)))
-		b = append(b, gaugeBytes...)
-	}
-
-	// Field 7: sum (message) - if used
-	if m.sum.dataPoint.timeNano != 0 {
-		b = appendProtoTag(b, 7, protoTypeLength)
-		sumBytes := m.sum.appendTo(nil)
-		b = binary.AppendUvarint(b, uint64(len(sumBytes)))
-		b = append(b, sumBytes...)
-	}
-	return b
-}
-
-func (g *otelGauge) appendTo(b []byte) []byte {
-	// Field 1: dataPoints (repeated message)
-	b = appendProtoTag(b, 1, protoTypeLength)
-	dataPointBytes := g.dataPoint.appendTo(nil)
-	b = binary.AppendUvarint(b, uint64(len(dataPointBytes)))
-	b = append(b, dataPointBytes...)
-
-	return b
-}
-
-func (s *otelSum) appendTo(b []byte) []byte {
-	// Field 1: dataPoints (repeated message)
-	b = appendProtoTag(b, 1, protoTypeLength)
-	dataPointBytes := s.dataPoint.appendTo(nil)
-	b = binary.AppendUvarint(b, uint64(len(dataPointBytes)))
-	b = append(b, dataPointBytes...)
-
-	// Field 2: aggregationTemporality (enum)
-	if s.aggregationTemporality != 0 {
-		b = appendProtoTag(b, 2, protoTypeVarint)
-		b = binary.AppendUvarint(b, uint64(s.aggregationTemporality))
-	}
-
-	// Field 3: isMonotonic (bool)
-	if s.isMonotonic {
-		b = appendProtoTag(b, 3, protoTypeVarint)
-		b = binary.AppendUvarint(b, 1) // true
-	}
-
-	return b
-}
-
-func (d *otelNumDataPoint) appendTo(b []byte) []byte {
-	// Field 2: startTimeUnixNano (fixed64)
-	if d.startNano != 0 {
-		b = appendProtoTag(b, 2, protoType64bit)
-		b = binary.LittleEndian.AppendUint64(b, uint64(d.startNano))
-	}
-
-	// Field 3: timeUnixNano (fixed64)
-	b = appendProtoTag(b, 3, protoType64bit)
-	b = binary.LittleEndian.AppendUint64(b, uint64(d.timeNano))
-
-	// Field 4: asDouble (double)
-	if d.vDouble != 0 {
-		b = appendProtoTag(b, 4, protoType64bit)
-		b = binary.LittleEndian.AppendUint64(b, math.Float64bits(d.vDouble))
-	}
-
-	// Field 6: asInt (sfixed64)
-	if d.vInt != 0 {
-		b = appendProtoTag(b, 6, protoType64bit)
-		b = binary.LittleEndian.AppendUint64(b, uint64(d.vInt))
-	}
-
-	// Field 7: attributes
-	b = appendOtelAttributesTo(b, 7, d.attributes)
-
-	return b
-}
-
-func appendOtelAttributesTo(b []byte, fieldNumber int, attrs map[string]any) []byte {
-outer:
-	for key, value := range attrs {
-		b = appendProtoTag(b, fieldNumber, protoTypeLength)
-
-		kvBytes := []byte{}
-		// Field 1: key (string)
-		kvBytes = appendProtoTag(kvBytes, 1, protoTypeLength)
-		kvBytes = appendProtoString(kvBytes, key)
-
-		// Field 2: value (AnyValue)
-		kvBytes = appendProtoTag(kvBytes, 2, protoTypeLength)
-
-		var anyValueBytes []byte
-		switch t := value.(type) {
-		case *string, string:
-			var v string
-			switch t := t.(type) {
-			case string:
-				v = t
-			case *string:
-				if t != nil {
-					v = *t
-				}
-			}
-			anyValueBytes = appendProtoTag(anyValueBytes, 1, protoTypeLength) // string_value = 1
-			anyValueBytes = appendProtoString(anyValueBytes, v)
-		case bool:
-			anyValueBytes = appendProtoTag(anyValueBytes, 2, protoTypeVarint) // bool_value = 2
-			if t {
-				anyValueBytes = binary.AppendUvarint(anyValueBytes, 1)
-			} else {
-				anyValueBytes = binary.AppendUvarint(anyValueBytes, 0)
-			}
-		case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, uintptr:
-			anyValueBytes = appendProtoTag(anyValueBytes, 3, protoTypeVarint) // int_value = 3
-			var v uint64
-			switch t := t.(type) {
-			case int:
-				v = uint64(t)
-			case int8:
-				v = uint64(t)
-			case int16:
-				v = uint64(t)
-			case int32:
-				v = uint64(t)
-			case int64:
-				v = uint64(t)
-			case uint:
-				v = uint64(t)
-			case uint8:
-				v = uint64(t)
-			case uint16:
-				v = uint64(t)
-			case uint32:
-				v = uint64(t)
-			case uint64:
-				v = t
-			case uintptr:
-				v = uint64(t)
-			}
-			anyValueBytes = binary.AppendUvarint(anyValueBytes, v)
-		case float32, float64:
-			var v float64
-			switch t := t.(type) {
-			case float32:
-				v = float64(t)
-			case float64:
-				v = t
-			}
-			anyValueBytes = appendProtoTag(anyValueBytes, 4, protoType64bit) // double_value = 4
-			anyValueBytes = binary.LittleEndian.AppendUint64(anyValueBytes, math.Float64bits(v))
-		case []byte:
-			anyValueBytes = appendProtoTag(anyValueBytes, 7, protoTypeLength) // bytes_value = 7
-			anyValueBytes = binary.AppendUvarint(anyValueBytes, uint64(len(t)))
-			anyValueBytes = append(anyValueBytes, t...)
-		default:
-			continue outer
-		}
-
-		kvBytes = binary.AppendUvarint(kvBytes, uint64(len(anyValueBytes)))
-		kvBytes = append(kvBytes, anyValueBytes...)
-
-		b = binary.AppendUvarint(b, uint64(len(kvBytes)))
-		b = append(b, kvBytes...)
-	}
-	return b
-}
diff --git a/pkg/kgo/producer.go b/pkg/kgo/producer.go
index 26307994..25ef7ca4 100644
--- a/pkg/kgo/producer.go
+++ b/pkg/kgo/producer.go
@@ -14,6 +14,8 @@ import (
 )
 
 type producer struct {
+	inflight atomicI64 // high 16: # waiters, low 48: # inflight
+
 	// mu and c are used for flush and drain notifications; mu is used for
 	// a few other tight locks.
 	mu sync.Mutex
@@ -59,9 +61,23 @@ type producer struct {
 	idVersion int16
 
 	batchPromises ringBatchPromise
+	promisesMu    sync.Mutex
 
 	txnMu sync.Mutex
 	inTxn bool
+
+	// If using EndBeginTxnUnsafe, and any partitions are actually produced
+	// to, we issue an AddPartitionsToTxn at the end to re-add them to a
+	// new transaction. We have to due to logic races: the broker may not
+	// have handled the produce requests yet, and we want to ensure a new
+	// transaction is started.
+	//
+	// If the user stops producing, we want to ensure that our restarted
+	// transaction is actually ended. Thus, we set readded whenever we have
+	// partitions we actually restart. We issue EndTxn and reset readded in
+	// EndAndBegin; if nothing more was produced to, we ensure we finish
+	// the started txn.
+	readded bool
 }
 
 // BufferedProduceRecords returns the number of records currently buffered for
@@ -85,83 +101,6 @@ func (cl *Client) BufferedProduceBytes() int64 {
 	return cl.producer.bufferedBytes + cl.producer.blockedBytes
 }
 
-// EnsureProduceConnectionIsOpen attempts to open a produce connection to all
-// specified brokers, or all brokers if `brokers` is empty or contains -1.
-//
-// This can be used in an attempt to reduce the latency when producing if your
-// application produces infrequently: you can force open a produce connection a
-// bit earlier than you intend to produce, rather than at the moment you
-// produce. In rare circumstances, it is possible that a connection that was
-// ensured to be open may close before you produce.
-//
-// This returns an errors.Join'd error that merges a message for all brokers
-// that failed to be opened as well as why.
-func (cl *Client) EnsureProduceConnectionIsOpen(ctx context.Context, brokers ...int32) error {
-	var (
-		keep = brokers[:0]
-		all  bool
-		wg   sync.WaitGroup
-		mu   sync.Mutex
-		errs []error
-	)
-	for _, b := range brokers {
-		switch {
-		case b < -1:
-		case b == -1:
-			all = true
-		case b > -1:
-			keep = append(keep, b)
-		}
-	}
-	var toOpen []*broker
-	if all || len(brokers) == 0 {
-		cl.brokersMu.RLock()
-		toOpen = cl.brokers
-		if len(toOpen) == 0 {
-			cl.brokersMu.RUnlock()
-			if err := cl.fetchBrokerMetadata(ctx); err != nil {
-				return err
-			}
-			cl.brokersMu.RLock()
-			toOpen = cl.brokers
-		}
-		cl.brokersMu.RUnlock()
-	} else {
-		for _, b := range brokers {
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-
-				br, err := cl.brokerOrErr(ctx, b, errUnknownBroker)
-
-				mu.Lock()
-				defer mu.Unlock()
-				if err != nil {
-					errs = append(errs, fmt.Errorf("%d: %w", b, err))
-					return
-				}
-				toOpen = append(toOpen, br)
-			}()
-		}
-		wg.Wait()
-	}
-
-	for _, br := range toOpen {
-		wg.Add(1)
-		br.do(ctx, &forceOpenReq{new(kmsg.ProduceRequest)}, func(_ kmsg.Response, err error) {
-			defer wg.Done()
-			if err != nil {
-				mu.Lock()
-				errs = append(errs, fmt.Errorf("%d: %w", br.meta.NodeID, err))
-				mu.Unlock()
-			}
-		})
-	}
-	wg.Wait()
-
-	return errors.Join(errs...)
-}
-
 type unknownTopicProduces struct {
 	buffered []promisedRec
 	wait     chan error // retryable errors
@@ -617,6 +556,7 @@ func (p *producer) finishPromises(b batchPromise) {
 		}
 	}()
 start:
+	p.promisesMu.Lock()
 	for i, pr := range b.recs {
 		pr.LeaderEpoch = 0
 		if b.baseOffset == -1 {
@@ -634,6 +574,7 @@ start:
 		broadcast = broadcast || recBroadcast
 		b.recs[i] = promisedRec{}
 	}
+	p.promisesMu.Unlock()
 	if cap(b.recs) > 4 {
 		cl.prsPool.put(b.recs)
 	}
@@ -1188,6 +1129,60 @@ func (cl *Client) Flush(ctx context.Context) error {
 	}
 }
 
+func (p *producer) pause(ctx context.Context) error {
+	p.inflight.Add(1 << 48)
+
+	quit := false
+	done := make(chan struct{})
+	go func() {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+		defer close(done)
+		for !quit && p.inflight.Load()&((1<<48)-1) != 0 {
+			p.c.Wait()
+		}
+	}()
+
+	select {
+	case <-done:
+		return nil
+	case <-ctx.Done():
+		p.mu.Lock()
+		quit = true
+		p.mu.Unlock()
+		p.c.Broadcast()
+		p.resume() // dec our inflight
+		return ctx.Err()
+	}
+}
+
+func (p *producer) resume() {
+	if p.inflight.Add(-1<<48) == 0 {
+		p.cl.allSinksAndSources(func(sns sinkAndSource) {
+			sns.sink.maybeDrain()
+		})
+	}
+}
+
+func (p *producer) maybeAddInflight() bool {
+	if p.inflight.Load()>>48 > 0 {
+		return false
+	}
+	if p.inflight.Add(1)>>48 > 0 {
+		p.decInflight()
+		return false
+	}
+	return true
+}
+
+func (p *producer) decInflight() {
+	if p.inflight.Add(-1)>>48 > 0 {
+		p.mu.Lock()
+		p.mu.Unlock() //nolint:gocritic,staticcheck // We use the lock as a barrier, unlocking immediately is safe.
+		p.c.Broadcast()
+	}
+}
+
 // Bumps the tries for all buffered records in the client.
 //
 // This is called whenever there is a problematic error that would affect the
diff --git a/pkg/kgo/sink.go b/pkg/kgo/sink.go
index b291d295..591f8351 100644
--- a/pkg/kgo/sink.go
+++ b/pkg/kgo/sink.go
@@ -399,6 +399,15 @@ func (s *sink) produce(sem <-chan struct{}) bool {
 		return false
 	}
 
+	if !s.cl.producer.maybeAddInflight() { // must do before marking recBufs on a txn
+		return false
+	}
+	defer func() {
+		if !produced {
+			s.cl.producer.decInflight()
+		}
+	}()
+
 	// NOTE: we create the req AFTER getting our producer ID!
 	//
 	// If a prior response caused errReloadProducerID, then calling
@@ -469,6 +478,7 @@ func (s *sink) produce(sem <-chan struct{}) bool {
 	batches := req.batches.sliced()
 	s.doSequenced(req, func(br *broker, resp kmsg.Response, err error) {
 		s.handleReqResp(br, req, resp, err)
+		s.cl.producer.decInflight()
 		batches.eachOwnerLocked((*recBatch).decInflight)
 		<-sem
 	})
@@ -848,11 +858,6 @@ func (s *sink) handleReqRespBatch(
 	err := kerr.ErrorForCode(rp.ErrorCode)
 	failUnknown := batch.owner.checkUnknownFailLimit(err)
 	switch {
-	case err == kerr.ConcurrentTransactions:
-		// Occasionally this is bubbled back to the producer as of
-		// KIP-890; we retry this.
-		fallthrough //nolint:gocritic // easier to read this way
-
 	case kerr.IsRetriable(err) &&
 		!failUnknown &&
 		err != kerr.CorruptMessage &&
@@ -2308,7 +2313,7 @@ func (b seqRecBatch) appendTo(
 		defer byteBuffers.Put(w)
 		w.Reset()
 
-		compressed, codec := compressor.Compress(w, toCompress, mkCompressFlags(version)...)
+		compressed, codec := compressor.Compress(w, toCompress, version)
 		if compressed != nil && // nil would be from an error
 			len(compressed) < len(toCompress) {
 			// our compressed was shorter: copy over
@@ -2390,7 +2395,7 @@ func (b seqRecBatch) appendToAsMessageSet(dst []byte, version uint8, compressor
 		defer byteBuffers.Put(w)
 		w.Reset()
 
-		compressed, codec := compressor.Compress(w, toCompress, mkCompressFlags(int16(version))...)
+		compressed, codec := compressor.Compress(w, toCompress, int16(version))
 		inner := &Record{Value: compressed}
 		wrappedLength := messageSet0Length(inner)
 		if version == 2 {
diff --git a/pkg/kgo/topics_and_partitions.go b/pkg/kgo/topics_and_partitions.go
index ca8746e2..e35b90f6 100644
--- a/pkg/kgo/topics_and_partitions.go
+++ b/pkg/kgo/topics_and_partitions.go
@@ -1,6 +1,7 @@
 package kgo
 
 import (
+	"cmp"
 	"encoding/hex"
 	"fmt"
 	"slices"
@@ -83,6 +84,23 @@ func mapi32sDeepEq(l, r map[string][]int32) bool {
 	return true
 }
 
+func slicesDeepEq[S ~[]E, E cmp.Ordered](l, r S) bool {
+	if len(l) != len(r) {
+		return false
+	}
+	if l == nil && r != nil {
+		return false
+	}
+	if r == nil && l != nil {
+		return false
+	}
+	dupl := slices.Clone(l)
+	dupr := slices.Clone(r)
+	slices.Sort(dupl)
+	slices.Sort(dupr)
+	return slices.Equal(dupl, dupr)
+}
+
 func (a *amtps) store(m map[string][]int32) { a.v.Store(m) }
 
 type mtps map[string][]int32
diff --git a/pkg/kgo/txn.go b/pkg/kgo/txn.go
index 0c6a4646..486375f9 100644
--- a/pkg/kgo/txn.go
+++ b/pkg/kgo/txn.go
@@ -522,37 +522,133 @@ const (
 	EndBeginTxnUnsafe
 )
 
-// EndAndBeginTransaction is a combination of Flush, EndTransaction, and
-// BeginTransaction. You cannot concurrently produce during this function.
+// EndAndBeginTransaction is a combination of EndTransaction and
+// BeginTransaction, and relaxes the restriction that the client must have no
+// buffered records. This function does not flush nor abort any buffered
+// records. It is ok to concurrently produce while this function executes.
 //
 // The onEnd function is called with your input context and the result of
-// Flush, or EndTransaction if Flush is successful. If onEnd returns an error,
-// BeginTransaction is not called and this function returns the result of
-// onEnd. Otherwise, this function returns the result of BeginTransaction. See
-// the documentation on EndTransaction and BeginTransaction for further
-// details. It is invalid to call this function more than once at a time, and
-// it is invalid to call concurrent with EndTransaction or BeginTransaction.
-//
-// This function used to serve more purpose, allowing you to produce
-// concurrently while calling this and avoiding flushing, but the internal
-// optimizations are no longer valid as of Kafka 4.0 due to KIP-890 changing
-// some internal semantics.
+// EndTransaction. Promises are paused while onEnd executes. If onEnd returns
+// an error, BeginTransaction is not called and this function returns the
+// result of onEnd. Otherwise, this function returns the result of
+// BeginTransaction. See the documentation on EndTransaction and
+// BeginTransaction for further details. It is invalid to call this function
+// more than once at a time, and it is invalid to call concurrent with
+// EndTransaction or BeginTransaction.
 func (cl *Client) EndAndBeginTransaction(
 	ctx context.Context,
 	_ EndBeginTxnHow,
 	commit TransactionEndTry,
 	onEnd func(context.Context, error) error,
 ) (rerr error) {
+	if g := cl.consumer.g; g != nil {
+		return errors.New("cannot use EndAndBeginTransaction with EOS")
+	}
+
+	cl.producer.txnMu.Lock()
+	defer cl.producer.txnMu.Unlock()
+
+	// From BeginTransaction: if we return with no error, we begin.  Unlike
+	// BeginTransaction, we do not error if in a transaction, because we
+	// expect to be in one.
 	defer func() {
-		rerr = onEnd(ctx, rerr)
 		if rerr == nil {
-			rerr = cl.BeginTransaction()
+			needRecover, didRecover, err := cl.maybeRecoverProducerID(ctx)
+			if needRecover && !didRecover {
+				cl.cfg.logger.Log(LogLevelInfo, "unable to begin transaction due to unrecoverable producer id error", "err", err)
+				rerr = fmt.Errorf("producer ID has a fatal, unrecoverable error, err: %w", err)
+				return
+			}
+			cl.producer.inTxn = true
+			cl.cfg.logger.Log(LogLevelInfo, "beginning transaction", "transactional_id", *cl.cfg.txnID)
 		}
 	}()
-	if err := cl.Flush(ctx); err != nil {
+
+	// We have to pause AddPartitionsToTxn and ProduceRequest, and we only
+	// resume after the user's onEnd has been called.
+	if err := cl.producer.pause(ctx); err != nil {
 		return err
 	}
-	return cl.EndTransaction(ctx, commit)
+	defer cl.producer.resume()
+
+	// Before BeginTransaction, we block promises & call onEnd with whatever
+	// the return error is.
+	cl.producer.promisesMu.Lock()
+	var promisesUnblocked bool
+	unblockPromises := func() {
+		if promisesUnblocked {
+			return
+		}
+		promisesUnblocked = true
+		defer cl.producer.promisesMu.Unlock()
+		rerr = onEnd(ctx, rerr)
+	}
+	defer unblockPromises()
+
+	if !cl.producer.inTxn {
+		return nil
+	}
+
+	var anyAdded bool
+	for _, parts := range cl.producer.topics.load() {
+		for _, part := range parts.load().partitions {
+			if part.records.addedToTxn.Swap(false) {
+				anyAdded = true
+			}
+		}
+	}
+	anyAdded = anyAdded || cl.producer.readded
+
+	// EndTxn when no txn was started returns INVALID_TXN_STATE.
+	if !anyAdded {
+		cl.cfg.logger.Log(LogLevelDebug, "no records were produced during the commit; thus no transaction was began; ending without doing anything")
+		return nil
+	}
+
+	// From EndTransaction: if the pid has an error, we may try to recover.
+	id, epoch, err := cl.producerID(ctx2fn(ctx))
+	if err != nil {
+		if commit {
+			return kerr.OperationNotAttempted
+		}
+		if _, didRecover, _ := cl.maybeRecoverProducerID(ctx); didRecover {
+			return nil
+		}
+	}
+	cl.cfg.logger.Log(LogLevelInfo, "ending transaction",
+		"transactional_id", *cl.cfg.txnID,
+		"producer_id", id,
+		"epoch", epoch,
+		"commit", commit,
+	)
+	cl.producer.readded = false
+	err = cl.doWithConcurrentTransactions(ctx, "EndTxn", func() error {
+		req := kmsg.NewPtrEndTxnRequest()
+		req.TransactionalID = *cl.cfg.txnID
+		req.ProducerID = id
+		req.ProducerEpoch = epoch
+		req.Commit = bool(commit)
+		resp, err := req.RequestWith(ctx, cl)
+		if err != nil {
+			return err
+		}
+		if err = kerr.ErrorForCode(resp.ErrorCode); err != nil {
+			return err
+		}
+		if resp.Version >= 5 && resp.ProducerID >= 0 {
+			cl.producer.id.Store(&producerID{
+				id:    resp.ProducerID,
+				epoch: resp.ProducerEpoch,
+			})
+			cl.resetAllProducerSequences()
+		}
+		return nil
+	})
+	var ke *kerr.Error
+	if errors.As(err, &ke) && !ke.Retriable {
+		cl.failProducerID(id, epoch, err)
+	}
+	return err
 }
 
 // AbortBufferedRecords fails all unflushed records with ErrAborted and waits
@@ -679,6 +775,8 @@ func (cl *Client) EndTransaction(ctx context.Context, commit TransactionEndTry)
 		}
 	}
 
+	anyAdded = anyAdded || cl.producer.readded
+
 	// If no partition was added to a transaction, then we have nothing to commit.
 	//
 	// Note that anyAdded is true if the producer ID was failed, meaning we will
@@ -706,11 +804,12 @@ func (cl *Client) EndTransaction(ctx context.Context, commit TransactionEndTry)
 
 	cl.cfg.logger.Log(LogLevelInfo, "ending transaction",
 		"transactional_id", *cl.cfg.txnID,
-		"commit", commit,
 		"producer_id", id,
 		"epoch", epoch,
+		"commit", commit,
 	)
 
+	cl.producer.readded = false
 	err = cl.doWithConcurrentTransactions(ctx, "EndTxn", func() error {
 		req := kmsg.NewPtrEndTxnRequest()
 		req.TransactionalID = *cl.cfg.txnID
@@ -730,21 +829,6 @@ func (cl *Client) EndTransaction(ctx context.Context, commit TransactionEndTry)
 				epoch: resp.ProducerEpoch,
 			})
 			cl.resetAllProducerSequences()
-			cl.cfg.logger.Log(LogLevelInfo, "end transaction response successfully received",
-				"transactional_id", *cl.cfg.txnID,
-				"commit", commit,
-				"prior_id", id,
-				"prior_epoch", epoch,
-				"new_id", resp.ProducerID,
-				"new_epoch", resp.ProducerEpoch,
-			)
-		} else {
-			cl.cfg.logger.Log(LogLevelInfo, "end transaction response successfully received",
-				"transactional_id", *cl.cfg.txnID,
-				"commit", commit,
-				"producer_id", id,
-				"epoch", epoch,
-			)
 		}
 		return nil
 	})
@@ -1041,13 +1125,11 @@ func (g *groupConsumer) commitTxn(ctx context.Context, req *kmsg.TxnOffsetCommit
 		var resp *kmsg.TxnOffsetCommitResponse
 		var err error
 		if len(req.Topics) > 0 {
-			start := time.Now()
 			resp, err = req.RequestWith(commitCtx, g.cl)
-			if err != nil {
-				onDone(req, nil, err)
-				return
-			}
-			g.cl.metrics.observeTime(&g.cl.metrics.cCommitLatency, time.Since(start).Milliseconds())
+		}
+		if err != nil {
+			onDone(req, nil, err)
+			return
 		}
 		onDone(req, resp, nil)
 	}()
diff --git a/pkg/sr/client.go b/pkg/sr/client.go
index a436255d..4a893aea 100644
--- a/pkg/sr/client.go
+++ b/pkg/sr/client.go
@@ -19,6 +19,7 @@ package sr
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -55,6 +56,7 @@ func (e *ResponseError) Error() string {
 type Client struct {
 	urls      []string
 	httpcl    *http.Client
+	dialTLS   *tls.Config
 	ua        string
 	defParams Param
 	opts      []ClientOpt
@@ -84,6 +86,32 @@ func NewClient(opts ...ClientOpt) (*Client, error) {
 		return nil, errors.New("unable to create client with no URLs")
 	}
 
+	if cl.httpcl == nil {
+		return nil, errors.New("unable to create client with no http client")
+	}
+
+	if cl.dialTLS != nil {
+		// Sanity check some invariants: if the user is using DialTLSConfig
+		// and a custom http client, the client must be compatible with us
+		// setting the one needed field (and no others should be set).
+		if cl.httpcl.Transport != nil {
+			tr, ok := cl.httpcl.Transport.(*http.Transport)
+			if !ok {
+				return nil, errors.New("unable to use DialTLSConfig with a custom http.Client that does not use an *http.Transport")
+			}
+			if tr.Dial != nil || tr.DialContext != nil || tr.DialTLS != nil || tr.DialTLSContext != nil {
+				return nil, errors.New("unable to use DialTLSConfig with an http.Client that has a custom dial function")
+			}
+			tr = tr.Clone()
+			tr.TLSClientConfig = cl.dialTLS
+			cl.httpcl.Transport = tr
+		} else {
+			tr := http.DefaultTransport.(*http.Transport).Clone()
+			tr.TLSClientConfig = cl.dialTLS
+			cl.httpcl.Transport = tr
+		}
+	}
+
 	return cl, nil
 }
 
diff --git a/pkg/sr/clientopt.go b/pkg/sr/clientopt.go
index 39b31740..ca5435ef 100644
--- a/pkg/sr/clientopt.go
+++ b/pkg/sr/clientopt.go
@@ -2,10 +2,8 @@ package sr
 
 import (
 	"crypto/tls"
-	"net"
 	"net/http"
 	"strings"
-	"time"
 )
 
 type (
@@ -44,27 +42,10 @@ func URLs(urls ...string) ClientOpt {
 	}}
 }
 
-// DialTLSConfig sets a tls.Config to use in the default http client.
+// DialTLSConfig sets a tls.Config to use in the http client, either the default client or a client previously
+// set with the HTTPClient option. When setting this option, HTTP/2 support may not be enabled by default.
 func DialTLSConfig(c *tls.Config) ClientOpt {
-	return clientOpt{func(cl *Client) {
-		cl.httpcl = &http.Client{
-			Timeout: 5 * time.Second,
-			Transport: &http.Transport{
-				Proxy: http.ProxyFromEnvironment,
-				DialContext: (&net.Dialer{
-					Timeout:   30 * time.Second,
-					KeepAlive: 30 * time.Second,
-				}).DialContext,
-				TLSClientConfig:       c,
-				ForceAttemptHTTP2:     true,
-				MaxIdleConns:          100,
-				MaxIdleConnsPerHost:   100,
-				IdleConnTimeout:       90 * time.Second,
-				TLSHandshakeTimeout:   10 * time.Second,
-				ExpectContinueTimeout: 1 * time.Second,
-			},
-		}
-	}}
+	return clientOpt{func(cl *Client) { cl.dialTLS = c }}
 }
 
 // BasicAuth sets basic authorization to use for every request.
diff --git a/plugin/kprom/kprom.go b/plugin/kprom/kprom.go
index abbe9f23..a425dad9 100644
--- a/plugin/kprom/kprom.go
+++ b/plugin/kprom/kprom.go
@@ -400,6 +400,7 @@ func (m *Metrics) OnClientClosed(*kgo.Client) {
 			m.cfg.reg.Unregister(c)
 		}
 	}
+	m.cfg.reg = nil
 }
 
 // OnBrokerConnect implements the HookBrokerConnect interface for metrics


Test patch:
diff --git a/.github/workflows/lint-and-test.yml b/.github/workflows/lint-and-test.yml
index d7da9f79..ca4a7196 100644
--- a/.github/workflows/lint-and-test.yml
+++ b/.github/workflows/lint-and-test.yml
@@ -36,7 +36,7 @@ jobs:
     container: golang:latest
     steps:
       - uses: actions/checkout@v4
-      - run: cd pkg/kfake && go work init ../.. . && go test .
+      - run: cd pkg/kfake && go test .
 
   integration-test-kafka:
     if: github.repository == 'twmb/franz-go'
diff --git a/pkg/kfake/issues_test.go b/pkg/kfake/issues_test.go
index a07d0bd4..9ec2151c 100644
--- a/pkg/kfake/issues_test.go
+++ b/pkg/kfake/issues_test.go
@@ -2,9 +2,7 @@ package kfake
 
 import (
 	"context"
-	"os"
 	"strconv"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -91,6 +89,7 @@ func TestIssue885(t *testing.T) {
 		if req.Version < 11 {
 			t.Fatal("unable to run test with fetch requests < v11")
 		}
+
 		if len(req.Topics) != 1 || len(req.Topics[0].Partitions) != 1 {
 			t.Fatalf("unexpected malformed req topics or partitions: %v", req)
 		}
@@ -159,266 +158,3 @@ func TestIssue885(t *testing.T) {
 		consumed += fs.NumRecords()
 	}
 }
-
-func TestIssue905(t *testing.T) {
-	const (
-		testTopic        = "foo"
-		producedMessages = 5
-	)
-
-	c, err := NewCluster(
-		NumBrokers(2),
-		SleepOutOfOrder(),
-		SeedTopics(1, testTopic),
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer c.Close()
-
-	// Flow:
-	//
-	// * Leader always returns the follower
-	//
-	// * We produce 5 separate batches just to have some data
-	//
-	// END SETUP STAGE.
-	//
-	// TEST
-	//
-	// * We set recheck period to 1s
-	// * We fetch -- leader should be hit once, then follower
-	// * We sleep >1s before polling again. Follower should be hit a second time (buffering a fetch), then redirect to leader, then follower.
-	//
-	// If we do not redirect back to follower within 2s, consider failure.
-
-	// Inline anonymous function so that we can defer and cleanup within scope.
-	func() {
-		cl, err := kgo.NewClient(
-			kgo.DefaultProduceTopic(testTopic),
-			kgo.SeedBrokers(c.ListenAddrs()...),
-		)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer cl.Close()
-
-		for i := 0; i < producedMessages; i++ {
-			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-			err := cl.ProduceSync(ctx, kgo.StringRecord(strconv.Itoa(i))).FirstErr()
-			cancel()
-			if err != nil {
-				t.Fatal(err)
-			}
-		}
-	}()
-
-	ti := c.TopicInfo(testTopic)
-	pi := c.PartitionInfo(testTopic, 0)
-	follower := (pi.Leader + 1) % 2
-	c.SetFollowers(testTopic, 0, []int32{follower})
-
-	var leaderReqs, followerReqs atomic.Int32
-	allowFollower := make(chan struct{}, 1)
-	c.ControlKey(1, func(kreq kmsg.Request) (kmsg.Response, error, bool) {
-		c.KeepControl()
-
-		req := kreq.(*kmsg.FetchRequest)
-		if req.Version < 11 {
-			t.Fatal("unable to run test with fetch requests < v11")
-		}
-		if len(req.Topics) != 1 || len(req.Topics[0].Partitions) != 1 {
-			t.Fatalf("unexpected malformed req topics or partitions: %v", req)
-		}
-
-		req.MaxBytes = 1 // Always ensure only one batch is returned
-
-		// Every leader request we redirect back to the follower.
-		if c.CurrentNode() == pi.Leader {
-			leaderReqs.Add(1)
-
-			resp := req.ResponseKind().(*kmsg.FetchResponse)
-			rt := kmsg.NewFetchResponseTopic()
-			rt.Topic = testTopic
-			rt.TopicID = ti.TopicID
-			rp := kmsg.NewFetchResponseTopicPartition()
-
-			resp.Topics = append(resp.Topics, rt)
-			rtp := &resp.Topics[0]
-
-			rtp.Partitions = append(rtp.Partitions, rp)
-			rpp := &rtp.Partitions[0]
-
-			rpp.Partition = 0
-			rpp.ErrorCode = 0
-			rpp.HighWatermark = pi.HighWatermark
-			rpp.LastStableOffset = pi.LastStableOffset
-			rpp.LogStartOffset = 0
-
-			rpp.PreferredReadReplica = (pi.Leader + 1) % 2
-			return resp, nil, true
-		}
-
-		<-allowFollower
-		followerReqs.Add(1)
-
-		return nil, nil, false
-	})
-
-	cl, err := kgo.NewClient(
-		kgo.SeedBrokers(c.ListenAddrs()...),
-		kgo.ConsumeTopics(testTopic),
-		kgo.Rack("foo"),
-		kgo.DisableFetchSessions(),
-		kgo.RecheckPreferredReplicaInterval(time.Second),
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer cl.Close()
-
-	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
-	defer cancel()
-
-	chkfs := func(fs kgo.Fetches) {
-		if fs.NumRecords() != 1 {
-			t.Errorf("got %d records != exp 1", fs.NumRecords())
-		}
-		if len(fs.Errors()) != 0 {
-			t.Errorf("got fetch errors: %v", fs.Errors())
-		}
-	}
-
-	// First poll. Standard; triggers us to fetch from the follower.
-	// We guard if the broker can reply to ensure our followerReqs
-	// check does not race against a client internally buffering
-	// another fetch quickly.
-	{
-		allowFollower <- struct{}{}
-		fs := cl.PollFetches(ctx)
-		chkfs(fs)
-		if lr := leaderReqs.Load(); lr != 1 {
-			t.Errorf("stage 1 leader reqs %d != exp 1", lr)
-		}
-		if fr := followerReqs.Load(); fr != 1 {
-			t.Errorf("stage 1 follower reqs reqs %d != exp 1", fr)
-		}
-		allowFollower <- struct{}{} // allow a background buffered fetch
-	}
-
-	// Sleep 1s. We are now polling a buffered fetch from the follower.
-	time.Sleep(time.Second)
-	for followerReqs.Load() != 2 {
-		time.Sleep(50 * time.Millisecond)
-	}
-	{
-		fs := cl.PollFetches(ctx)
-		chkfs(fs)
-		if lr := leaderReqs.Load(); lr != 1 {
-			t.Errorf("stage 2 leader reqs %d != exp 1", lr)
-		}
-		if fr := followerReqs.Load(); fr != 2 {
-			t.Errorf("stage 2 follower reqs reqs %d != exp 2", fr)
-		}
-		allowFollower <- struct{}{} // again allow a background buffered fetch; this one will notice we need to redir back to follower
-	}
-
-	// Poll again. This is buffered, and the NEXT one should go back
-	// to the leader again.
-	{
-		fs := cl.PollFetches(ctx)
-		chkfs(fs)
-		if lr := leaderReqs.Load(); lr != 1 {
-			t.Errorf("stage 3 leader reqs %d != exp 1", lr)
-		}
-		if fr := followerReqs.Load(); fr != 3 {
-			t.Errorf("stage 3 follower reqs reqs %d != exp 3", fr)
-		}
-		close(allowFollower) // allow all reqs; the next check is our last
-	}
-
-	// We now expect leader finally before the follower again.
-	{
-		fs := cl.PollFetches(ctx)
-		chkfs(fs)
-		if lr := leaderReqs.Load(); lr != 2 {
-			t.Errorf("stage 4 leader reqs %d != exp 2", lr)
-		}
-		if fr := followerReqs.Load(); fr != 4 {
-			t.Errorf("stage 4 follower reqs reqs %d != exp 4", fr)
-		}
-	}
-
-	// Success.
-}
-
-func TestIssue906(t *testing.T) {
-	const testTopic = "foo"
-
-	c, err := NewCluster(
-		NumBrokers(1),
-		SleepOutOfOrder(),
-		SeedTopics(1, testTopic),
-		AllowAutoTopicCreation(),
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer c.Close()
-
-	// Seed "foo" with two records.
-	func() {
-		cl, err := kgo.NewClient(
-			kgo.DefaultProduceTopic(testTopic),
-			kgo.SeedBrokers(c.ListenAddrs()...),
-		)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer cl.Close()
-
-		for i := 0; i < 2; i++ {
-			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-			err := cl.ProduceSync(ctx, kgo.StringRecord(strconv.Itoa(i))).FirstErr()
-			cancel()
-			if err != nil {
-				t.Fatal(err)
-			}
-		}
-	}()
-
-	client, err := kgo.NewClient(
-		kgo.SeedBrokers(c.ListenAddrs()...),
-		kgo.ConsumeTopics("^foo.*"),
-		kgo.ConsumeRegex(),
-		kgo.AllowAutoTopicCreation(),
-		kgo.WithLogger(kgo.BasicLogger(os.Stderr, kgo.LogLevelDebug, nil)),
-	)
-	if err != nil {
-		t.Fatal(err)
-		return
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	defer cancel()
-
-	var produced int
-	for produced != 2 {
-		fetches := client.PollRecords(ctx, 10)
-
-		var records []*kgo.Record
-		iter := fetches.RecordIter()
-		for !iter.Done() {
-			record := iter.Next()
-			records = append(records, &kgo.Record{
-				Topic: "bar",
-				Key:   record.Key,
-				Value: record.Value,
-			})
-		}
-
-		if err := client.ProduceSync(ctx, records...).FirstErr(); err != nil {
-			t.Errorf("unable to produce: %v", err)
-		}
-		produced += len(records)
-	}
-}
diff --git a/pkg/kgo/compression_test.go b/pkg/kgo/compression_test.go
index cc2ef65a..dd55f232 100644
--- a/pkg/kgo/compression_test.go
+++ b/pkg/kgo/compression_test.go
@@ -79,11 +79,11 @@ func TestCompressDecompress(t *testing.T) {
 	}
 
 	var wg sync.WaitGroup
-	for _, flag := range []CompressFlag{
-		0, CompressDisableZstd,
+	for _, produceVersion := range []int16{
+		0, 7,
 	} {
 		wg.Add(1)
-		go func(flag CompressFlag) {
+		go func(produceVersion int16) {
 			defer wg.Done()
 			for _, codecs := range [][]CompressionCodec{
 				{{codec: 0}},
@@ -109,7 +109,7 @@ func TestCompressDecompress(t *testing.T) {
 						for _, in := range inputs {
 							w.Reset()
 
-							got, used := c.Compress(w, in, flag)
+							got, used := c.Compress(w, in, produceVersion)
 							got, err := d.Decompress(got, used)
 							if err != nil {
 								t.Errorf("unexpected decompress err: %v", err)
@@ -122,7 +122,7 @@ func TestCompressDecompress(t *testing.T) {
 					}()
 				}
 			}
-		}(flag)
+		}(produceVersion)
 	}
 	wg.Wait()
 }
diff --git a/pkg/kgo/helpers_test.go b/pkg/kgo/helpers_test.go
index 6d6995b2..aaf7566f 100644
--- a/pkg/kgo/helpers_test.go
+++ b/pkg/kgo/helpers_test.go
@@ -271,8 +271,8 @@ issue:
 	// If we run tests in a container _immediately_ after the container
 	// starts, we can receive dial errors for a bit if the container is not
 	// fully initialized. Handle this by retrying specifically dial errors.
-	if ne := (*net.OpError)(nil); errors.As(err, &ne) && ne.Op == "dial" && time.Since(start) < 30*time.Second {
-		time.Sleep(time.Second)
+	if ne := (*net.OpError)(nil); errors.As(err, &ne) && ne.Op == "dial" && time.Since(start) < 15*time.Second {
+		time.Sleep(10 * time.Millisecond)
 		goto issue
 	}
 
@@ -314,7 +314,7 @@ issue:
 			err = kerr.ErrorForCode(resp.Topics[0].ErrorCode)
 		}
 		if err != nil {
-			tb.Logf("unable to delete topic %q: %v", topic, err)
+			tb.Fatalf("unable to delete topic %q: %v", topic, err)
 		}
 	}
 }
@@ -587,7 +587,7 @@ out:
 		sort.Ints(allKeys)
 		for i := 0; i < testRecordLimit; i++ {
 			if allKeys[i] != i {
-				t.Errorf("consumers %d: got key %d != exp %d, first 100: %v", level, allKeys[i], i, allKeys[:100])
+				t.Fatalf("consumers %d: got key %d != exp %d, first 100: %v", level, allKeys[i], i, allKeys[:100])
 			}
 		}
 	}
diff --git a/pkg/kgo/produce_request_test.go b/pkg/kgo/produce_request_test.go
index 3bbf13a7..269c2dbd 100644
--- a/pkg/kgo/produce_request_test.go
+++ b/pkg/kgo/produce_request_test.go
@@ -388,7 +388,7 @@ func TestRecBatchAppendTo(t *testing.T) {
 		kbatch.Attributes |= 0x0002 // snappy
 		w := byteBuffers.Get().(*bytes.Buffer)
 		w.Reset()
-		kbatch.Records, _ = compressor.Compress(w, kbatch.Records)
+		kbatch.Records, _ = compressor.Compress(w, kbatch.Records, version)
 	}
 
 	fixFields()
diff --git a/pkg/kgo/txn_test.go b/pkg/kgo/txn_test.go
index 7471676f..3292a0ba 100644
--- a/pkg/kgo/txn_test.go
+++ b/pkg/kgo/txn_test.go
@@ -138,13 +138,12 @@ func (c *testConsumer) goTransact(txnsBeforeQuit int) {
 func (c *testConsumer) transact(txnsBeforeQuit int) {
 	defer c.wg.Done()
 
-	txid := randsha()
 	opts := []Opt{
 		// Kraft sometimes returns success from topic creation, and
 		// then returns UnknownTopicXyz for a while in metadata loads.
 		// It also returns NotLeaderXyz; we handle both problems.
 		UnknownTopicRetries(-1),
-		TransactionalID(txid),
+		TransactionalID(randsha()),
 		TransactionTimeout(60 * time.Second),
 		WithLogger(testLogger()),
 		// Control records have their own unique offset, so for testing,
@@ -178,7 +177,7 @@ func (c *testConsumer) transact(txnsBeforeQuit int) {
 			if consumed := int(c.consumed.Load()); consumed == testRecordLimit {
 				return
 			} else if consumed > testRecordLimit {
-				panic(fmt.Sprintf("invalid: consumed too much from %s (at %d, group %s, tx %s)", c.consumeFrom, consumed, c.group, txid))
+				panic("invalid: consumed too much")
 			}
 			continue
 		}
diff --git a/pkg/sr/clientopt_test.go b/pkg/sr/clientopt_test.go
new file mode 100644
index 00000000..6c97da73
--- /dev/null
+++ b/pkg/sr/clientopt_test.go
@@ -0,0 +1,90 @@
+package sr
+
+import (
+	"crypto/tls"
+	"net"
+	"net/http"
+	"testing"
+)
+
+func TestDialTLSConfigOpt(t *testing.T) {
+	t.Run("Defaults", func(t *testing.T) {
+		tlscfg := &tls.Config{} //nolint:gosec // not relevant for this test case
+
+		rcl, err := NewClient(DialTLSConfig(tlscfg))
+		if err != nil {
+			t.Fatalf("unable to create client: %s", err)
+		}
+
+		if rcl.dialTLS != tlscfg {
+			t.Errorf("TLS config: expected %+v, got %+v", tlscfg, rcl.dialTLS)
+		}
+
+		tr := rcl.httpcl.Transport.(*http.Transport)
+		if tr.TLSClientConfig != tlscfg {
+			t.Errorf("TLS config in transport: expected %+v, got %+v", tlscfg, tr.TLSClientConfig)
+		}
+	})
+
+	t.Run("CustomHTTPClient", func(t *testing.T) {
+		httpcl := &http.Client{}
+		tlscfg := &tls.Config{} //nolint:gosec // not relevant for this test case
+
+		rcl, err := NewClient(
+			HTTPClient(httpcl),
+			DialTLSConfig(tlscfg),
+		)
+		if err != nil {
+			t.Fatalf("unable to create client: %s", err)
+		}
+
+		if rcl.dialTLS != tlscfg {
+			t.Errorf("TLS config: expected %+v, got %+v", tlscfg, rcl.dialTLS)
+		}
+
+		tr := rcl.httpcl.Transport.(*http.Transport)
+		if tr.TLSClientConfig != tlscfg {
+			t.Errorf("TLS config in transport: expected %+v, got %+v", tlscfg, tr.TLSClientConfig)
+		}
+	})
+
+	t.Run("CustomHTTPTransport", func(t *testing.T) {
+		httptr := &http.Transport{}
+		httpcl := &http.Client{Transport: httptr}
+		tlscfg := &tls.Config{} //nolint:gosec // not relevant for this test case
+
+		rcl, err := NewClient(
+			HTTPClient(httpcl),
+			DialTLSConfig(tlscfg),
+		)
+		if err != nil {
+			t.Fatalf("unable to create client: %s", err)
+		}
+
+		if rcl.dialTLS != tlscfg {
+			t.Errorf("TLS config: expected %+v, got %+v", tlscfg, rcl.dialTLS)
+		}
+
+		tr := rcl.httpcl.Transport.(*http.Transport)
+		if tr.TLSClientConfig != tlscfg {
+			t.Errorf("TLS config in transport: expected %+v, got %+v", tlscfg, tr.TLSClientConfig)
+		}
+	})
+
+	t.Run("IncompatibleHTTPClient", func(t *testing.T) {
+		dialFn := func(_, _ string) (net.Conn, error) { return nil, nil }
+		httptr := &http.Transport{Dial: dialFn}
+		httpcl := &http.Client{Transport: httptr}
+		tlscfg := &tls.Config{} //nolint:gosec // not relevant for this test case
+
+		_, err := NewClient(
+			HTTPClient(httpcl),
+			DialTLSConfig(tlscfg),
+		)
+
+		expected := "unable to use DialTLSConfig with an http.Client that has a custom dial function"
+		if err == nil || err.Error() != expected {
+			t.Errorf("new client error: expected %q, got %q", expected, err)
+		}
+	})
+}