Repository: Shopify/sarama
Pull request #2378 (2022-11-06 06:34:18)

fix(producer): return errors for every message in retryBatch to avoid producer hang forever
This fixes https://github.com/Shopify/sarama/issues/2377

cch123: CLA signed
dnwe: @cch123 do you think you might be able to add a functional test that covers this scenario?

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bc1221b..818d8ff 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -8,7 +8,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go-version: [1.18.x, 1.19.x]
+        go-version: [1.19.x]
     steps:
     - uses: actions/checkout@v3
     - name: Setup Go
@@ -40,8 +40,8 @@ jobs:
     - name: Get Go environment
       id: go-env
       run: |
-        echo "::set-output name=cache::$(go env GOCACHE)"
-        echo "::set-output name=modcache::$(go env GOMODCACHE)"
+        echo "cache=$(go env GOCACHE)" >> $GITHUB_OUTPUT
+        echo "modcache=$(go env GOMODCACHE)" >> $GITHUB_OUTPUT
     - name: Set up cache
       uses: actions/cache@v3
       with:
diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml
index 0901c26..fd418c4 100644
--- a/.github/workflows/fuzz.yml
+++ b/.github/workflows/fuzz.yml
@@ -25,8 +25,8 @@ jobs:
     - name: Get Go environment
       id: go-env
       run: |
-        echo "::set-output name=cache::$(go env GOCACHE)"
-        echo "::set-output name=modcache::$(go env GOMODCACHE)"
+        echo "cache=$(go env GOCACHE)" >> $GITHUB_OUTPUT
+        echo "modcache=$(go env GOMODCACHE)" >> $GITHUB_OUTPUT
 
     - name: Set up cache
       uses: actions/cache@v3
diff --git a/.github/workflows/fvt.yml b/.github/workflows/fvt.yml
index ab5e6c7..c252021 100644
--- a/.github/workflows/fvt.yml
+++ b/.github/workflows/fvt.yml
@@ -13,7 +13,7 @@ jobs:
       fail-fast: false
       matrix:
         go-version: [1.19.x]
-        kafka-version: [2.8.2, 3.0.2, 3.1.2, 3.2.3]
+        kafka-version: [2.8.2, 3.1.2, 3.2.3, 3.3.1]
     env:
       DEBUG: true
       GOFLAGS: -trimpath
@@ -27,8 +27,8 @@ jobs:
     - name: Get Go environment
       id: go-env
       run: |
-        echo "::set-output name=cache::$(go env GOCACHE)"
-        echo "::set-output name=modcache::$(go env GOMODCACHE)"
+        echo "cache=$(go env GOCACHE)" >> $GITHUB_OUTPUT
+        echo "modcache=$(go env GOMODCACHE)" >> $GITHUB_OUTPUT
     - name: Set up cache
       uses: actions/cache@v3
       with:
diff --git a/Dockerfile.kafka b/Dockerfile.kafka
index c361f6d..48a9c17 100644
--- a/Dockerfile.kafka
+++ b/Dockerfile.kafka
@@ -15,12 +15,12 @@ RUN cd /etc/java/java-11-openjdk/*/conf/security \
   && echo 'networkaddress.cache.ttl=0' >> java.security \
   && echo 'networkaddress.cache.negative.ttl=0' >> java.security
 
-# https://github.com/apache/kafka/blob/0d518aaed158896ee9ee6949b8f38128d1d73634/tests/docker/Dockerfile#L65-L67
+# https://github.com/apache/kafka/blob/53eeaad946cd053e9eb1a762972d4efeacb8e4fc/tests/docker/Dockerfile#L65-L69
 ARG KAFKA_MIRROR="https://s3-us-west-2.amazonaws.com/kafka-packages"
 RUN mkdir -p "/opt/kafka-2.8.2" && chmod a+rw /opt/kafka-2.8.2 && curl -s "$KAFKA_MIRROR/kafka_2.12-2.8.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-2.8.2"
-RUN mkdir -p "/opt/kafka-3.0.2" && chmod a+rw /opt/kafka-3.0.2 && curl -s "$KAFKA_MIRROR/kafka_2.12-3.0.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-3.0.2"
 RUN mkdir -p "/opt/kafka-3.1.2" && chmod a+rw /opt/kafka-3.1.2 && curl -s "$KAFKA_MIRROR/kafka_2.12-3.1.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-3.1.2"
 RUN mkdir -p "/opt/kafka-3.2.3" && chmod a+rw /opt/kafka-3.2.3 && curl -s "$KAFKA_MIRROR/kafka_2.12-3.2.3.tgz" | tar xz --strip-components=1 -C "/opt/kafka-3.2.3"
+RUN mkdir -p "/opt/kafka-3.3.1" && chmod a+rw /opt/kafka-3.3.1 && curl -s "$KAFKA_MIRROR/kafka_2.12-3.3.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-3.3.1"
 
 COPY entrypoint.sh /
 
diff --git a/admin.go b/admin.go
index f06198a..a334daf 100644
--- a/admin.go
+++ b/admin.go
@@ -280,17 +280,7 @@ func (ca *clusterAdmin) DescribeTopics(topics []string) (metadata []*TopicMetada
 		return nil, err
 	}
 
-	request := &MetadataRequest{
-		Topics:                 topics,
-		AllowAutoTopicCreation: false,
-	}
-
-	if ca.conf.Version.IsAtLeast(V1_0_0_0) {
-		request.Version = 5
-	} else if ca.conf.Version.IsAtLeast(V0_11_0_0) {
-		request.Version = 4
-	}
-
+	request := NewMetadataRequest(ca.conf.Version, topics)
 	response, err := controller.GetMetadata(request)
 	if err != nil {
 		return nil, err
@@ -304,14 +294,7 @@ func (ca *clusterAdmin) DescribeCluster() (brokers []*Broker, controllerID int32
 		return nil, int32(0), err
 	}
 
-	request := &MetadataRequest{
-		Topics: []string{},
-	}
-
-	if ca.conf.Version.IsAtLeast(V0_10_0_0) {
-		request.Version = 1
-	}
-
+	request := NewMetadataRequest(ca.conf.Version, nil)
 	response, err := controller.GetMetadata(request)
 	if err != nil {
 		return nil, int32(0), err
@@ -352,7 +335,7 @@ func (ca *clusterAdmin) ListTopics() (map[string]TopicDetail, error) {
 	}
 	_ = b.Open(ca.client.Config())
 
-	metadataReq := &MetadataRequest{}
+	metadataReq := NewMetadataRequest(ca.conf.Version, nil)
 	metadataResp, err := b.GetMetadata(metadataReq)
 	if err != nil {
 		return nil, err
diff --git a/async_producer.go b/async_producer.go
index 5c23ac7..50f226f 100644
--- a/async_producer.go
+++ b/async_producer.go
@@ -613,6 +613,18 @@ func (pp *partitionProducer) backoff(retries int) {
 	}
 }
 
+func (pp *partitionProducer) updateLeaderIfBrokerProducerIsNil(msg *ProducerMessage) error {
+	if pp.brokerProducer == nil {
+		if err := pp.updateLeader(); err != nil {
+			pp.parent.returnError(msg, err)
+			pp.backoff(msg.retries)
+			return err
+		}
+		Logger.Printf("producer/leader/%s/%d selected broker %d\n", pp.topic, pp.partition, pp.leader.ID())
+	}
+	return nil
+}
+
 func (pp *partitionProducer) dispatch() {
 	// try to prefetch the leader; if this doesn't work, we'll do a proper call to `updateLeader`
 	// on the first message
@@ -644,6 +656,9 @@ func (pp *partitionProducer) dispatch() {
 		}
 
 		if msg.retries > pp.highWatermark {
+			if err := pp.updateLeaderIfBrokerProducerIsNil(msg); err != nil {
+				continue
+			}
 			// a new, higher, retry level; handle it and then back off
 			pp.newHighWatermark(msg.retries)
 			pp.backoff(msg.retries)
@@ -670,14 +685,8 @@ func (pp *partitionProducer) dispatch() {
 
 		// if we made it this far then the current msg contains real data, and can be sent to the next goroutine
 		// without breaking any of our ordering guarantees
-
-		if pp.brokerProducer == nil {
-			if err := pp.updateLeader(); err != nil {
-				pp.parent.returnError(msg, err)
-				pp.backoff(msg.retries)
-				continue
-			}
-			Logger.Printf("producer/leader/%s/%d selected broker %d\n", pp.topic, pp.partition, pp.leader.ID())
+		if err := pp.updateLeaderIfBrokerProducerIsNil(msg); err != nil {
+			continue
 		}
 
 		// Now that we know we have a broker to actually try and send this message to, generate the sequence
@@ -1152,7 +1161,7 @@ func (p *asyncProducer) retryBatch(topic string, partition int32, pSet *partitio
 	produceSet.bufferCount += len(pSet.msgs)
 	for _, msg := range pSet.msgs {
 		if msg.retries >= p.conf.Producer.Retry.Max {
-			p.returnError(msg, kerr)
+			p.returnErrors(pSet.msgs, kerr)
 			return
 		}
 		msg.retries++
diff --git a/broker.go b/broker.go
index 5815fd5..b9f0db5 100644
--- a/broker.go
+++ b/broker.go
@@ -33,26 +33,28 @@ type Broker struct {
 	responses     chan *responsePromise
 	done          chan bool
 
-	metricRegistry         metrics.Registry
-	incomingByteRate       metrics.Meter
-	requestRate            metrics.Meter
-	fetchRate              metrics.Meter
-	requestSize            metrics.Histogram
-	requestLatency         metrics.Histogram
-	outgoingByteRate       metrics.Meter
-	responseRate           metrics.Meter
-	responseSize           metrics.Histogram
-	requestsInFlight       metrics.Counter
-	brokerIncomingByteRate metrics.Meter
-	brokerRequestRate      metrics.Meter
-	brokerFetchRate        metrics.Meter
-	brokerRequestSize      metrics.Histogram
-	brokerRequestLatency   metrics.Histogram
-	brokerOutgoingByteRate metrics.Meter
-	brokerResponseRate     metrics.Meter
-	brokerResponseSize     metrics.Histogram
-	brokerRequestsInFlight metrics.Counter
-	brokerThrottleTime     metrics.Histogram
+	metricRegistry             metrics.Registry
+	incomingByteRate           metrics.Meter
+	requestRate                metrics.Meter
+	fetchRate                  metrics.Meter
+	requestSize                metrics.Histogram
+	requestLatency             metrics.Histogram
+	outgoingByteRate           metrics.Meter
+	responseRate               metrics.Meter
+	responseSize               metrics.Histogram
+	requestsInFlight           metrics.Counter
+	protocolRequestsRate       map[int16]metrics.Meter
+	brokerIncomingByteRate     metrics.Meter
+	brokerRequestRate          metrics.Meter
+	brokerFetchRate            metrics.Meter
+	brokerRequestSize          metrics.Histogram
+	brokerRequestLatency       metrics.Histogram
+	brokerOutgoingByteRate     metrics.Meter
+	brokerResponseRate         metrics.Meter
+	brokerResponseSize         metrics.Histogram
+	brokerRequestsInFlight     metrics.Counter
+	brokerThrottleTime         metrics.Histogram
+	brokerProtocolRequestsRate map[int16]metrics.Meter
 
 	kerberosAuthenticator               GSSAPIKerberosAuth
 	clientSessionReauthenticationTimeMs int64
@@ -218,6 +220,7 @@ func (b *Broker) Open(conf *Config) error {
 		b.responseRate = metrics.GetOrRegisterMeter("response-rate", b.metricRegistry)
 		b.responseSize = getOrRegisterHistogram("response-size", b.metricRegistry)
 		b.requestsInFlight = metrics.GetOrRegisterCounter("requests-in-flight", b.metricRegistry)
+		b.protocolRequestsRate = map[int16]metrics.Meter{}
 		// Do not gather metrics for seeded broker (only used during bootstrap) because they share
 		// the same id (-1) and are already exposed through the global metrics above
 		if b.id >= 0 && !metrics.UseNilMetrics {
@@ -998,6 +1001,7 @@ func (b *Broker) sendInternal(rb protocolBody, promise *responsePromise) error {
 	b.addRequestInFlightMetrics(1)
 	bytes, err := b.write(buf)
 	b.updateOutgoingCommunicationMetrics(bytes)
+	b.updateProtocolMetrics(rb)
 	if err != nil {
 		b.addRequestInFlightMetrics(-1)
 		return err
@@ -1606,6 +1610,24 @@ func (b *Broker) updateOutgoingCommunicationMetrics(bytes int) {
 	}
 }
 
+func (b *Broker) updateProtocolMetrics(rb protocolBody) {
+	protocolRequestsRate := b.protocolRequestsRate[rb.key()]
+	if protocolRequestsRate == nil {
+		protocolRequestsRate = metrics.GetOrRegisterMeter(fmt.Sprintf("protocol-requests-rate-%d", rb.key()), b.metricRegistry)
+		b.protocolRequestsRate[rb.key()] = protocolRequestsRate
+	}
+	protocolRequestsRate.Mark(1)
+
+	if b.brokerProtocolRequestsRate != nil {
+		brokerProtocolRequestsRate := b.brokerProtocolRequestsRate[rb.key()]
+		if brokerProtocolRequestsRate == nil {
+			brokerProtocolRequestsRate = b.registerMeter(fmt.Sprintf("protocol-requests-rate-%d", rb.key()))
+			b.brokerProtocolRequestsRate[rb.key()] = brokerProtocolRequestsRate
+		}
+		brokerProtocolRequestsRate.Mark(1)
+	}
+}
+
 func (b *Broker) updateThrottleMetric(throttleTime time.Duration) {
 	if throttleTime != time.Duration(0) {
 		DebugLogger.Printf(
@@ -1629,6 +1651,7 @@ func (b *Broker) registerMetrics() {
 	b.brokerResponseSize = b.registerHistogram("response-size")
 	b.brokerRequestsInFlight = b.registerCounter("requests-in-flight")
 	b.brokerThrottleTime = b.registerHistogram("throttle-time-in-ms")
+	b.brokerProtocolRequestsRate = map[int16]metrics.Meter{}
 }
 
 func (b *Broker) registerMeter(name string) metrics.Meter {
diff --git a/client.go b/client.go
index b3b7dfb..f7872a1 100644
--- a/client.go
+++ b/client.go
@@ -50,6 +50,10 @@ type Client interface {
 	// topic/partition, as determined by querying the cluster metadata.
 	Leader(topic string, partitionID int32) (*Broker, error)
 
+	// LeaderAndEpoch returns the the leader and its epoch for the current
+	// topic/partition, as determined by querying the cluster metadata.
+	LeaderAndEpoch(topic string, partitionID int32) (*Broker, int32, error)
+
 	// Replicas returns the set of all replica IDs for the given partition.
 	Replicas(topic string, partitionID int32) ([]int32, error)
 
@@ -452,21 +456,25 @@ func (client *client) OfflineReplicas(topic string, partitionID int32) ([]int32,
 }
 
 func (client *client) Leader(topic string, partitionID int32) (*Broker, error) {
+	leader, _, err := client.LeaderAndEpoch(topic, partitionID)
+	return leader, err
+}
+
+func (client *client) LeaderAndEpoch(topic string, partitionID int32) (*Broker, int32, error) {
 	if client.Closed() {
-		return nil, ErrClosedClient
+		return nil, -1, ErrClosedClient
 	}
 
-	leader, err := client.cachedLeader(topic, partitionID)
-
+	leader, epoch, err := client.cachedLeader(topic, partitionID)
 	if leader == nil {
 		err = client.RefreshMetadata(topic)
 		if err != nil {
-			return nil, err
+			return nil, -1, err
 		}
-		leader, err = client.cachedLeader(topic, partitionID)
+		leader, epoch, err = client.cachedLeader(topic, partitionID)
 	}
 
-	return leader, err
+	return leader, epoch, err
 }
 
 func (client *client) RefreshBrokers(addrs []string) error {
@@ -848,7 +856,7 @@ func (client *client) setPartitionCache(topic string, partitionSet partitionType
 	return ret
 }
 
-func (client *client) cachedLeader(topic string, partitionID int32) (*Broker, error) {
+func (client *client) cachedLeader(topic string, partitionID int32) (*Broker, int32, error) {
 	client.lock.RLock()
 	defer client.lock.RUnlock()
 
@@ -857,18 +865,18 @@ func (client *client) cachedLeader(topic string, partitionID int32) (*Broker, er
 		metadata, ok := partitions[partitionID]
 		if ok {
 			if errors.Is(metadata.Err, ErrLeaderNotAvailable) {
-				return nil, ErrLeaderNotAvailable
+				return nil, -1, ErrLeaderNotAvailable
 			}
 			b := client.brokers[metadata.Leader]
 			if b == nil {
-				return nil, ErrLeaderNotAvailable
+				return nil, -1, ErrLeaderNotAvailable
 			}
 			_ = b.Open(client.conf)
-			return b, nil
+			return b, metadata.LeaderEpoch, nil
 		}
 	}
 
-	return nil, ErrUnknownTopicOrPartition
+	return nil, -1, ErrUnknownTopicOrPartition
 }
 
 func (client *client) getOffset(topic string, partitionID int32, time int64) (int64, error) {
@@ -989,13 +997,8 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 			DebugLogger.Printf("client/metadata fetching metadata for all topics from broker %s\n", broker.addr)
 		}
 
-		req := &MetadataRequest{Topics: topics, AllowAutoTopicCreation: allowAutoTopicCreation}
-		if client.conf.Version.IsAtLeast(V1_0_0_0) {
-			req.Version = 5
-		} else if client.conf.Version.IsAtLeast(V0_10_0_0) {
-			req.Version = 1
-		}
-
+		req := NewMetadataRequest(client.conf.Version, topics)
+		req.AllowAutoTopicCreation = allowAutoTopicCreation
 		t := atomic.LoadInt64(&client.updateMetaDataMs)
 		if !atomic.CompareAndSwapInt64(&client.updateMetaDataMs, t, time.Now().UnixNano()/int64(time.Millisecond)) {
 			return nil
diff --git a/consumer.go b/consumer.go
index d635dfe..eb27df8 100644
--- a/consumer.go
+++ b/consumer.go
@@ -165,6 +165,7 @@ func (c *consumer) ConsumePartition(topic string, partition int32, offset int64)
 		messages:             make(chan *ConsumerMessage, c.conf.ChannelBufferSize),
 		errors:               make(chan *ConsumerError, c.conf.ChannelBufferSize),
 		feeder:               make(chan *FetchResponse, 1),
+		leaderEpoch:          invalidLeaderEpoch,
 		preferredReadReplica: invalidPreferredReplicaID,
 		trigger:              make(chan none, 1),
 		dying:                make(chan none),
@@ -175,9 +176,8 @@ func (c *consumer) ConsumePartition(topic string, partition int32, offset int64)
 		return nil, err
 	}
 
-	var leader *Broker
-	var err error
-	if leader, err = c.client.Leader(child.topic, child.partition); err != nil {
+	leader, epoch, err := c.client.LeaderAndEpoch(child.topic, child.partition)
+	if err != nil {
 		return nil, err
 	}
 
@@ -188,6 +188,7 @@ func (c *consumer) ConsumePartition(topic string, partition int32, offset int64)
 	go withRecover(child.dispatcher)
 	go withRecover(child.responseFeeder)
 
+	child.leaderEpoch = epoch
 	child.broker = c.refBrokerConsumer(leader)
 	child.broker.input <- child
 
@@ -400,6 +401,7 @@ type partitionConsumer struct {
 	errors   chan *ConsumerError
 	feeder   chan *FetchResponse
 
+	leaderEpoch          int32
 	preferredReadReplica int32
 
 	trigger, dying chan none
@@ -463,11 +465,11 @@ func (child *partitionConsumer) dispatcher() {
 	close(child.feeder)
 }
 
-func (child *partitionConsumer) preferredBroker() (*Broker, error) {
+func (child *partitionConsumer) preferredBroker() (*Broker, int32, error) {
 	if child.preferredReadReplica >= 0 {
 		broker, err := child.consumer.client.Broker(child.preferredReadReplica)
 		if err == nil {
-			return broker, nil
+			return broker, child.leaderEpoch, nil
 		}
 		Logger.Printf(
 			"consumer/%s/%d failed to find active broker for preferred read replica %d - will fallback to leader",
@@ -480,7 +482,7 @@ func (child *partitionConsumer) preferredBroker() (*Broker, error) {
 	}
 
 	// if preferred replica cannot be found fallback to leader
-	return child.consumer.client.Leader(child.topic, child.partition)
+	return child.consumer.client.LeaderAndEpoch(child.topic, child.partition)
 }
 
 func (child *partitionConsumer) dispatch() error {
@@ -488,13 +490,13 @@ func (child *partitionConsumer) dispatch() error {
 		return err
 	}
 
-	broker, err := child.preferredBroker()
+	broker, epoch, err := child.preferredBroker()
 	if err != nil {
 		return err
 	}
 
+	child.leaderEpoch = epoch
 	child.broker = child.consumer.refBrokerConsumer(broker)
-
 	child.broker.input <- child
 
 	return nil
@@ -987,7 +989,7 @@ func (bc *brokerConsumer) handleResponses() {
 		child.responseResult = nil
 
 		if result == nil {
-			if preferredBroker, err := child.preferredBroker(); err == nil {
+			if preferredBroker, _, err := child.preferredBroker(); err == nil {
 				if bc.broker.ID() != preferredBroker.ID() {
 					// not an error but needs redispatching to consume from preferred replica
 					Logger.Printf(
@@ -1014,7 +1016,12 @@ func (bc *brokerConsumer) handleResponses() {
 			Logger.Printf("consumer/%s/%d shutting down because %s\n", child.topic, child.partition, result)
 			close(child.trigger)
 			delete(bc.subscriptions, child)
-		} else if errors.Is(result, ErrUnknownTopicOrPartition) || errors.Is(result, ErrNotLeaderForPartition) || errors.Is(result, ErrLeaderNotAvailable) || errors.Is(result, ErrReplicaNotAvailable) {
+		} else if errors.Is(result, ErrUnknownTopicOrPartition) ||
+			errors.Is(result, ErrNotLeaderForPartition) ||
+			errors.Is(result, ErrLeaderNotAvailable) ||
+			errors.Is(result, ErrReplicaNotAvailable) ||
+			errors.Is(result, ErrFencedLeaderEpoch) ||
+			errors.Is(result, ErrUnknownLeaderEpoch) {
 			// not an error, but does need redispatching
 			Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because %s\n",
 				bc.broker.ID(), child.topic, child.partition, result)
@@ -1092,7 +1099,7 @@ func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
 
 	for child := range bc.subscriptions {
 		if !child.IsPaused() {
-			request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize)
+			request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize, child.leaderEpoch)
 		}
 	}
 
diff --git a/docker-compose.yml b/docker-compose.yml
index 5d3f038..e1119c8 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -40,7 +40,7 @@ services:
       dockerfile: Dockerfile.kafka
     restart: always
     environment:
-      KAFKA_VERSION: ${KAFKA_VERSION:-3.1.2}
+      KAFKA_VERSION: ${KAFKA_VERSION:-3.3.1}
       KAFKA_CFG_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
       KAFKA_CFG_LISTENERS: 'LISTENER_INTERNAL://:9091,LISTENER_LOCAL://:29091'
       KAFKA_CFG_ADVERTISED_LISTENERS: 'LISTENER_INTERNAL://kafka-1:9091,LISTENER_LOCAL://localhost:29091'
@@ -62,7 +62,7 @@ services:
       dockerfile: Dockerfile.kafka
     restart: always
     environment:
-      KAFKA_VERSION: ${KAFKA_VERSION:-3.1.2}
+      KAFKA_VERSION: ${KAFKA_VERSION:-3.3.1}
       KAFKA_CFG_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
       KAFKA_CFG_LISTENERS: 'LISTENER_INTERNAL://:9091,LISTENER_LOCAL://:29092'
       KAFKA_CFG_ADVERTISED_LISTENERS: 'LISTENER_INTERNAL://kafka-2:9091,LISTENER_LOCAL://localhost:29092'
@@ -84,7 +84,7 @@ services:
       dockerfile: Dockerfile.kafka
     restart: always
     environment:
-      KAFKA_VERSION: ${KAFKA_VERSION:-3.1.2}
+      KAFKA_VERSION: ${KAFKA_VERSION:-3.3.1}
       KAFKA_CFG_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
       KAFKA_CFG_LISTENERS: 'LISTENER_INTERNAL://:9091,LISTENER_LOCAL://:29093'
       KAFKA_CFG_ADVERTISED_LISTENERS: 'LISTENER_INTERNAL://kafka-3:9091,LISTENER_LOCAL://localhost:29093'
@@ -106,7 +106,7 @@ services:
       dockerfile: Dockerfile.kafka
     restart: always
     environment:
-      KAFKA_VERSION: ${KAFKA_VERSION:-3.1.2}
+      KAFKA_VERSION: ${KAFKA_VERSION:-3.3.1}
       KAFKA_CFG_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
       KAFKA_CFG_LISTENERS: 'LISTENER_INTERNAL://:9091,LISTENER_LOCAL://:29094'
       KAFKA_CFG_ADVERTISED_LISTENERS: 'LISTENER_INTERNAL://kafka-4:9091,LISTENER_LOCAL://localhost:29094'
@@ -128,7 +128,7 @@ services:
       dockerfile: Dockerfile.kafka
     restart: always
     environment:
-      KAFKA_VERSION: ${KAFKA_VERSION:-3.1.2}
+      KAFKA_VERSION: ${KAFKA_VERSION:-3.3.1}
       KAFKA_CFG_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
       KAFKA_CFG_LISTENERS: 'LISTENER_INTERNAL://:9091,LISTENER_LOCAL://:29095'
       KAFKA_CFG_ADVERTISED_LISTENERS: 'LISTENER_INTERNAL://kafka-5:9091,LISTENER_LOCAL://localhost:29095'
diff --git a/entrypoint.sh b/entrypoint.sh
index cbcbcfc..8cd2efc 100755
--- a/entrypoint.sh
+++ b/entrypoint.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-KAFKA_VERSION="${KAFKA_VERSION:-3.1.2}"
+KAFKA_VERSION="${KAFKA_VERSION:-3.3.1}"
 KAFKA_HOME="/opt/kafka-${KAFKA_VERSION}"
 
 if [ ! -d "${KAFKA_HOME}" ]; then
diff --git a/examples/http_server/http_server.go b/examples/http_server/http_server.go
index d955d85..9209a52 100644
--- a/examples/http_server/http_server.go
+++ b/examples/http_server/http_server.go
@@ -16,13 +16,13 @@ import (
 )
 
 var (
-	addr      = flag.String("addr", ":8080", "The address to bind to")
-	brokers   = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The Kafka brokers to connect to, as a comma separated list")
-	verbose   = flag.Bool("verbose", false, "Turn on Sarama logging")
-	certFile  = flag.String("certificate", "", "The optional certificate file for client authentication")
-	keyFile   = flag.String("key", "", "The optional key file for client authentication")
-	caFile    = flag.String("ca", "", "The optional certificate authority file for TLS client authentication")
-	verifySsl = flag.Bool("verify", false, "Optional verify ssl certificates chain")
+	addr          = flag.String("addr", ":8080", "The address to bind to")
+	brokers       = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The Kafka brokers to connect to, as a comma separated list")
+	verbose       = flag.Bool("verbose", false, "Turn on Sarama logging")
+	certFile      = flag.String("certificate", "", "The optional certificate file for client authentication")
+	keyFile       = flag.String("key", "", "The optional key file for client authentication")
+	caFile        = flag.String("ca", "", "The optional certificate authority file for TLS client authentication")
+	tlsSkipVerify = flag.Bool("tls-skip-verify", false, "Whether to skip TLS server cert verification")
 )
 
 func main() {
@@ -71,7 +71,7 @@ func createTlsConfiguration() (t *tls.Config) {
 		t = &tls.Config{
 			Certificates:       []tls.Certificate{cert},
 			RootCAs:            caCertPool,
-			InsecureSkipVerify: *verifySsl,
+			InsecureSkipVerify: *tlsSkipVerify,
 		}
 	}
 	// will be nil by default if nothing is provided
diff --git a/examples/sasl_scram_client/main.go b/examples/sasl_scram_client/main.go
index e0ff84e..7c3221f 100644
--- a/examples/sasl_scram_client/main.go
+++ b/examples/sasl_scram_client/main.go
@@ -17,25 +17,25 @@ func init() {
 }
 
 var (
-	brokers   = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The Kafka brokers to connect to, as a comma separated list")
-	userName  = flag.String("username", "", "The SASL username")
-	passwd    = flag.String("passwd", "", "The SASL password")
-	algorithm = flag.String("algorithm", "", "The SASL SCRAM SHA algorithm sha256 or sha512 as mechanism")
-	topic     = flag.String("topic", "default_topic", "The Kafka topic to use")
-	certFile  = flag.String("certificate", "", "The optional certificate file for client authentication")
-	keyFile   = flag.String("key", "", "The optional key file for client authentication")
-	caFile    = flag.String("ca", "", "The optional certificate authority file for TLS client authentication")
-	verifySSL = flag.Bool("verify", false, "Optional verify ssl certificates chain")
-	useTLS    = flag.Bool("tls", false, "Use TLS to communicate with the cluster")
-	mode      = flag.String("mode", "produce", "Mode to run in: \"produce\" to produce, \"consume\" to consume")
-	logMsg    = flag.Bool("logmsg", false, "True to log consumed messages to console")
+	brokers       = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The Kafka brokers to connect to, as a comma separated list")
+	userName      = flag.String("username", "", "The SASL username")
+	passwd        = flag.String("passwd", "", "The SASL password")
+	algorithm     = flag.String("algorithm", "", "The SASL SCRAM SHA algorithm sha256 or sha512 as mechanism")
+	topic         = flag.String("topic", "default_topic", "The Kafka topic to use")
+	certFile      = flag.String("certificate", "", "The optional certificate file for client authentication")
+	keyFile       = flag.String("key", "", "The optional key file for client authentication")
+	caFile        = flag.String("ca", "", "The optional certificate authority file for TLS client authentication")
+	tlsSkipVerify = flag.Bool("tls-skip-verify", false, "Whether to skip TLS server cert verification")
+	useTLS        = flag.Bool("tls", false, "Use TLS to communicate with the cluster")
+	mode          = flag.String("mode", "produce", "Mode to run in: \"produce\" to produce, \"consume\" to consume")
+	logMsg        = flag.Bool("logmsg", false, "True to log consumed messages to console")
 
 	logger = log.New(os.Stdout, "[Producer] ", log.LstdFlags)
 )
 
 func createTLSConfiguration() (t *tls.Config) {
 	t = &tls.Config{
-		InsecureSkipVerify: *verifySSL,
+		InsecureSkipVerify: *tlsSkipVerify,
 	}
 	if *certFile != "" && *keyFile != "" && *caFile != "" {
 		cert, err := tls.LoadX509KeyPair(*certFile, *keyFile)
@@ -54,7 +54,7 @@ func createTLSConfiguration() (t *tls.Config) {
 		t = &tls.Config{
 			Certificates:       []tls.Certificate{cert},
 			RootCAs:            caCertPool,
-			InsecureSkipVerify: *verifySSL,
+			InsecureSkipVerify: *tlsSkipVerify,
 		}
 	}
 	return t
diff --git a/fetch_request.go b/fetch_request.go
index 63190a2..26adead 100644
--- a/fetch_request.go
+++ b/fetch_request.go
@@ -302,7 +302,7 @@ func (r *FetchRequest) requiredVersion() KafkaVersion {
 	}
 }
 
-func (r *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int64, maxBytes int32) {
+func (r *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int64, maxBytes int32, leaderEpoch int32) {
 	if r.blocks == nil {
 		r.blocks = make(map[string]map[int32]*fetchRequestBlock)
 	}
@@ -320,7 +320,7 @@ func (r *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int
 	tmp.maxBytes = maxBytes
 	tmp.fetchOffset = fetchOffset
 	if r.Version >= 9 {
-		tmp.currentLeaderEpoch = int32(-1)
+		tmp.currentLeaderEpoch = leaderEpoch
 	}
 
 	r.blocks[topic][partitionID] = tmp
diff --git a/fetch_response.go b/fetch_response.go
index c8ad604..3d449c8 100644
--- a/fetch_response.go
+++ b/fetch_response.go
@@ -8,7 +8,10 @@ import (
 	"github.com/rcrowley/go-metrics"
 )
 
-const invalidPreferredReplicaID = -1
+const (
+	invalidLeaderEpoch        = -1
+	invalidPreferredReplicaID = -1
+)
 
 type AbortedTransaction struct {
 	// ProducerID contains the producer id associated with the aborted transaction.
diff --git a/go.mod b/go.mod
index bdeb69e..9d7ba1b 100644
--- a/go.mod
+++ b/go.mod
@@ -17,7 +17,7 @@ require (
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
 	github.com/stretchr/testify v1.8.0
 	github.com/xdg-go/scram v1.1.1
-	golang.org/x/net v0.0.0-20220927171203-f486391704dc
+	golang.org/x/net v0.4.0
 	golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7
 )
 
@@ -33,7 +33,7 @@ require (
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/stringprep v1.0.3 // indirect
 	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa // indirect
-	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/text v0.5.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index c5cbb58..7005b6f 100644
--- a/go.sum
+++ b/go.sum
@@ -71,8 +71,8 @@ golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0
 golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220725212005-46097bf591d3/go.mod h1:AaygXjzTFtRAg2ttMY5RMuhpJ3cNnI0XpyFJD1iQRSM=
-golang.org/x/net v0.0.0-20220927171203-f486391704dc h1:FxpXZdoBqT8RjqTy6i1E8nXHhW21wK7ptQ/EPIGxzPQ=
-golang.org/x/net v0.0.0-20220927171203-f486391704dc/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
+golang.org/x/net v0.4.0 h1:Q5QPcMlvfxFTAPV0+07Xz/MpK9NTXu2VDUuy0FeMfaU=
+golang.org/x/net v0.4.0/go.mod h1:MBQ8lrhLObU/6UmLb4fmbmk5OcyYmqtbGd/9yIeKjEE=
 golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ffipvEDX2pl7E1TdqLqIc=
 golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -84,8 +84,9 @@ golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9sn
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.5.0 h1:OLmvp0KP+FVG99Ct/qFiL/Fhk4zp4QQnZ7b2U+5piUM=
+golang.org/x/text v0.5.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/metadata_request.go b/metadata_request.go
index e835f5a..a1b6ac0 100644
--- a/metadata_request.go
+++ b/metadata_request.go
@@ -1,13 +1,30 @@
 package sarama
 
 type MetadataRequest struct {
-	Version                int16
-	Topics                 []string
+	// Version defines the protocol version to use for encode and decode
+	Version int16
+	// Topics contains the topics to fetch metadata for.
+	Topics []string
+	// AllowAutoTopicCreation contains a If this is true, the broker may auto-create topics that we requested which do not already exist, if it is configured to do so.
 	AllowAutoTopicCreation bool
 }
 
-func (r *MetadataRequest) encode(pe packetEncoder) error {
-	if r.Version < 0 || r.Version > 5 {
+func NewMetadataRequest(version KafkaVersion, topics []string) *MetadataRequest {
+	m := &MetadataRequest{Topics: topics}
+	if version.IsAtLeast(V2_1_0_0) {
+		m.Version = 7
+	} else if version.IsAtLeast(V2_0_0_0) {
+		m.Version = 6
+	} else if version.IsAtLeast(V1_0_0_0) {
+		m.Version = 5
+	} else if version.IsAtLeast(V0_10_0_0) {
+		m.Version = 1
+	}
+	return m
+}
+
+func (r *MetadataRequest) encode(pe packetEncoder) (err error) {
+	if r.Version < 0 || r.Version > 12 {
 		return PacketEncodingError{"invalid or unsupported MetadataRequest version field"}
 	}
 	if r.Version == 0 || len(r.Topics) > 0 {
@@ -25,13 +42,15 @@ func (r *MetadataRequest) encode(pe packetEncoder) error {
 	} else {
 		pe.putInt32(-1)
 	}
-	if r.Version > 3 {
+
+	if r.Version >= 4 {
 		pe.putBool(r.AllowAutoTopicCreation)
 	}
+
 	return nil
 }
 
-func (r *MetadataRequest) decode(pd packetDecoder, version int16) error {
+func (r *MetadataRequest) decode(pd packetDecoder, version int16) (err error) {
 	r.Version = version
 	size, err := pd.getInt32()
 	if err != nil {
@@ -47,13 +66,13 @@ func (r *MetadataRequest) decode(pd packetDecoder, version int16) error {
 			r.Topics[i] = topic
 		}
 	}
-	if r.Version > 3 {
-		autoCreation, err := pd.getBool()
-		if err != nil {
+
+	if r.Version >= 4 {
+		if r.AllowAutoTopicCreation, err = pd.getBool(); err != nil {
 			return err
 		}
-		r.AllowAutoTopicCreation = autoCreation
 	}
+
 	return nil
 }
 
@@ -79,6 +98,10 @@ func (r *MetadataRequest) requiredVersion() KafkaVersion {
 		return V0_11_0_0
 	case 5:
 		return V1_0_0_0
+	case 6:
+		return V2_0_0_0
+	case 7:
+		return V2_1_0_0
 	default:
 		return MinVersion
 	}
diff --git a/metadata_response.go b/metadata_response.go
index 0bb8702..10a5687 100644
--- a/metadata_response.go
+++ b/metadata_response.go
@@ -1,44 +1,57 @@
 package sarama
 
+// PartitionMetadata contains each partition in the topic.
 type PartitionMetadata struct {
-	Err             KError
-	ID              int32
-	Leader          int32
-	Replicas        []int32
-	Isr             []int32
+	// Version defines the protocol version to use for encode and decode
+	Version int16
+	// Err contains the partition error, or 0 if there was no error.
+	Err KError
+	// ID contains the partition index.
+	ID int32
+	// Leader contains the ID of the leader broker.
+	Leader int32
+	// LeaderEpoch contains the leader epoch of this partition.
+	LeaderEpoch int32
+	// Replicas contains the set of all nodes that host this partition.
+	Replicas []int32
+	// Isr contains the set of nodes that are in sync with the leader for this partition.
+	Isr []int32
+	// OfflineReplicas contains the set of offline replicas of this partition.
 	OfflineReplicas []int32
 }
 
-func (pm *PartitionMetadata) decode(pd packetDecoder, version int16) (err error) {
+func (p *PartitionMetadata) decode(pd packetDecoder, version int16) (err error) {
+	p.Version = version
 	tmp, err := pd.getInt16()
 	if err != nil {
 		return err
 	}
-	pm.Err = KError(tmp)
+	p.Err = KError(tmp)
 
-	pm.ID, err = pd.getInt32()
-	if err != nil {
+	if p.ID, err = pd.getInt32(); err != nil {
 		return err
 	}
 
-	pm.Leader, err = pd.getInt32()
-	if err != nil {
+	if p.Leader, err = pd.getInt32(); err != nil {
 		return err
 	}
 
-	pm.Replicas, err = pd.getInt32Array()
-	if err != nil {
+	if p.Version >= 7 {
+		if p.LeaderEpoch, err = pd.getInt32(); err != nil {
+			return err
+		}
+	}
+
+	if p.Replicas, err = pd.getInt32Array(); err != nil {
 		return err
 	}
 
-	pm.Isr, err = pd.getInt32Array()
-	if err != nil {
+	if p.Isr, err = pd.getInt32Array(); err != nil {
 		return err
 	}
 
-	if version >= 5 {
-		pm.OfflineReplicas, err = pd.getInt32Array()
-		if err != nil {
+	if p.Version >= 5 {
+		if p.OfflineReplicas, err = pd.getInt32Array(); err != nil {
 			return err
 		}
 	}
@@ -46,24 +59,28 @@ func (pm *PartitionMetadata) decode(pd packetDecoder, version int16) (err error)
 	return nil
 }
 
-func (pm *PartitionMetadata) encode(pe packetEncoder, version int16) (err error) {
-	pe.putInt16(int16(pm.Err))
-	pe.putInt32(pm.ID)
-	pe.putInt32(pm.Leader)
+func (p *PartitionMetadata) encode(pe packetEncoder, version int16) (err error) {
+	p.Version = version
+	pe.putInt16(int16(p.Err))
 
-	err = pe.putInt32Array(pm.Replicas)
-	if err != nil {
+	pe.putInt32(p.ID)
+
+	pe.putInt32(p.Leader)
+
+	if p.Version >= 7 {
+		pe.putInt32(p.LeaderEpoch)
+	}
+
+	if err := pe.putInt32Array(p.Replicas); err != nil {
 		return err
 	}
 
-	err = pe.putInt32Array(pm.Isr)
-	if err != nil {
+	if err := pe.putInt32Array(p.Isr); err != nil {
 		return err
 	}
 
-	if version >= 5 {
-		err = pe.putInt32Array(pm.OfflineReplicas)
-		if err != nil {
+	if p.Version >= 5 {
+		if err := pe.putInt32Array(p.OfflineReplicas); err != nil {
 			return err
 		}
 	}
@@ -71,68 +88,71 @@ func (pm *PartitionMetadata) encode(pe packetEncoder, version int16) (err error)
 	return nil
 }
 
+// TopicMetadata contains each topic in the response.
 type TopicMetadata struct {
-	Err        KError
-	Name       string
-	IsInternal bool // Only valid for Version >= 1
+	// Version defines the protocol version to use for encode and decode
+	Version int16
+	// Err contains the topic error, or 0 if there was no error.
+	Err KError
+	// Name contains the topic name.
+	Name string
+	// IsInternal contains a True if the topic is internal.
+	IsInternal bool
+	// Partitions contains each partition in the topic.
 	Partitions []*PartitionMetadata
 }
 
-func (tm *TopicMetadata) decode(pd packetDecoder, version int16) (err error) {
+func (t *TopicMetadata) decode(pd packetDecoder, version int16) (err error) {
+	t.Version = version
 	tmp, err := pd.getInt16()
 	if err != nil {
 		return err
 	}
-	tm.Err = KError(tmp)
+	t.Err = KError(tmp)
 
-	tm.Name, err = pd.getString()
-	if err != nil {
+	if t.Name, err = pd.getString(); err != nil {
 		return err
 	}
 
-	if version >= 1 {
-		tm.IsInternal, err = pd.getBool()
-		if err != nil {
+	if t.Version >= 1 {
+		if t.IsInternal, err = pd.getBool(); err != nil {
 			return err
 		}
 	}
 
-	n, err := pd.getArrayLength()
-	if err != nil {
+	if numPartitions, err := pd.getArrayLength(); err != nil {
 		return err
-	}
-	tm.Partitions = make([]*PartitionMetadata, n)
-	for i := 0; i < n; i++ {
-		tm.Partitions[i] = new(PartitionMetadata)
-		err = tm.Partitions[i].decode(pd, version)
-		if err != nil {
-			return err
+	} else {
+		t.Partitions = make([]*PartitionMetadata, numPartitions)
+		for i := 0; i < numPartitions; i++ {
+			block := &PartitionMetadata{}
+			if err := block.decode(pd, t.Version); err != nil {
+				return err
+			}
+			t.Partitions[i] = block
 		}
 	}
 
 	return nil
 }
 
-func (tm *TopicMetadata) encode(pe packetEncoder, version int16) (err error) {
-	pe.putInt16(int16(tm.Err))
+func (t *TopicMetadata) encode(pe packetEncoder, version int16) (err error) {
+	t.Version = version
+	pe.putInt16(int16(t.Err))
 
-	err = pe.putString(tm.Name)
-	if err != nil {
+	if err := pe.putString(t.Name); err != nil {
 		return err
 	}
 
-	if version >= 1 {
-		pe.putBool(tm.IsInternal)
+	if t.Version >= 1 {
+		pe.putBool(t.IsInternal)
 	}
 
-	err = pe.putArrayLength(len(tm.Partitions))
-	if err != nil {
+	if err := pe.putArrayLength(len(t.Partitions)); err != nil {
 		return err
 	}
-
-	for _, pm := range tm.Partitions {
-		err = pm.encode(pe, version)
-		if err != nil {
+	for _, block := range t.Partitions {
+		if err := block.encode(pe, t.Version); err != nil {
 			return err
 		}
 	}
@@ -141,20 +161,24 @@ func (tm *TopicMetadata) encode(pe packetEncoder, version int16) (err error) {
 }
 
 type MetadataResponse struct {
-	Version        int16
+	// Version defines the protocol version to use for encode and decode
+	Version int16
+	// ThrottleTimeMs contains the duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota.
 	ThrottleTimeMs int32
-	Brokers        []*Broker
-	ClusterID      *string
-	ControllerID   int32
-	Topics         []*TopicMetadata
+	// Brokers contains each broker in the response.
+	Brokers []*Broker
+	// ClusterID contains the cluster ID that responding broker belongs to.
+	ClusterID *string
+	// ControllerID contains the ID of the controller broker.
+	ControllerID int32
+	// Topics contains each topic in the response.
+	Topics []*TopicMetadata
 }
 
 func (r *MetadataResponse) decode(pd packetDecoder, version int16) (err error) {
 	r.Version = version
-
-	if version >= 3 {
-		r.ThrottleTimeMs, err = pd.getInt32()
-		if err != nil {
+	if r.Version >= 3 {
+		if r.ThrottleTimeMs, err = pd.getInt32(); err != nil {
 			return err
 		}
 	}
@@ -163,7 +187,6 @@ func (r *MetadataResponse) decode(pd packetDecoder, version int16) (err error) {
 	if err != nil {
 		return err
 	}
-
 	r.Brokers = make([]*Broker, n)
 	for i := 0; i < n; i++ {
 		r.Brokers[i] = new(Broker)
@@ -173,46 +196,40 @@ func (r *MetadataResponse) decode(pd packetDecoder, version int16) (err error) {
 		}
 	}
 
-	if version >= 2 {
-		r.ClusterID, err = pd.getNullableString()
-		if err != nil {
+	if r.Version >= 2 {
+		if r.ClusterID, err = pd.getNullableString(); err != nil {
 			return err
 		}
 	}
 
-	if version >= 1 {
-		r.ControllerID, err = pd.getInt32()
-		if err != nil {
+	if r.Version >= 1 {
+		if r.ControllerID, err = pd.getInt32(); err != nil {
 			return err
 		}
-	} else {
-		r.ControllerID = -1
 	}
 
-	n, err = pd.getArrayLength()
-	if err != nil {
+	if numTopics, err := pd.getArrayLength(); err != nil {
 		return err
-	}
-
-	r.Topics = make([]*TopicMetadata, n)
-	for i := 0; i < n; i++ {
-		r.Topics[i] = new(TopicMetadata)
-		err = r.Topics[i].decode(pd, version)
-		if err != nil {
-			return err
+	} else {
+		r.Topics = make([]*TopicMetadata, numTopics)
+		for i := 0; i < numTopics; i++ {
+			block := &TopicMetadata{}
+			if err := block.decode(pd, r.Version); err != nil {
+				return err
+			}
+			r.Topics[i] = block
 		}
 	}
 
 	return nil
 }
 
-func (r *MetadataResponse) encode(pe packetEncoder) error {
+func (r *MetadataResponse) encode(pe packetEncoder) (err error) {
 	if r.Version >= 3 {
 		pe.putInt32(r.ThrottleTimeMs)
 	}
 
-	err := pe.putArrayLength(len(r.Brokers))
-	if err != nil {
+	if err := pe.putArrayLength(len(r.Brokers)); err != nil {
 		return err
 	}
 	for _, broker := range r.Brokers {
@@ -223,8 +240,7 @@ func (r *MetadataResponse) encode(pe packetEncoder) error {
 	}
 
 	if r.Version >= 2 {
-		err := pe.putNullableString(r.ClusterID)
-		if err != nil {
+		if err := pe.putNullableString(r.ClusterID); err != nil {
 			return err
 		}
 	}
@@ -233,13 +249,11 @@ func (r *MetadataResponse) encode(pe packetEncoder) error {
 		pe.putInt32(r.ControllerID)
 	}
 
-	err = pe.putArrayLength(len(r.Topics))
-	if err != nil {
+	if err := pe.putArrayLength(len(r.Topics)); err != nil {
 		return err
 	}
-	for _, tm := range r.Topics {
-		err = tm.encode(pe, r.Version)
-		if err != nil {
+	for _, block := range r.Topics {
+		if err := block.encode(pe, r.Version); err != nil {
 			return err
 		}
 	}
@@ -269,6 +283,10 @@ func (r *MetadataResponse) requiredVersion() KafkaVersion {
 		return V0_11_0_0
 	case 5:
 		return V1_0_0_0
+	case 6:
+		return V2_0_0_0
+	case 7:
+		return V2_1_0_0
 	default:
 		return MinVersion
 	}
@@ -316,7 +334,6 @@ func (r *MetadataResponse) AddTopicPartition(topic string, partition, brokerID i
 	tmatch.Partitions = append(tmatch.Partitions, pmatch)
 
 foundPartition:
-
 	pmatch.Leader = brokerID
 	pmatch.Replicas = replicas
 	pmatch.Isr = isr
diff --git a/request.go b/request.go
index 5fec977..1e3923d 100644
--- a/request.go
+++ b/request.go
@@ -125,7 +125,7 @@ func allocateBody(key, version int16) protocolBody {
 	case 2:
 		return &OffsetRequest{Version: version}
 	case 3:
-		return &MetadataRequest{}
+		return &MetadataRequest{Version: version}
 	case 8:
 		return &OffsetCommitRequest{Version: version}
 	case 9:
diff --git a/sarama.go b/sarama.go
index cb773f1..a42bc07 100644
--- a/sarama.go
+++ b/sarama.go
@@ -22,28 +22,32 @@ Metrics are exposed through https://github.com/rcrowley/go-metrics library in a
 
 Broker related metrics:
 
-	+----------------------------------------------+------------+---------------------------------------------------------------+
-	| Name                                         | Type       | Description                                                   |
-	+----------------------------------------------+------------+---------------------------------------------------------------+
-	| incoming-byte-rate                           | meter      | Bytes/second read off all brokers                             |
-	| incoming-byte-rate-for-broker-<broker-id>    | meter      | Bytes/second read off a given broker                          |
-	| outgoing-byte-rate                           | meter      | Bytes/second written off all brokers                          |
-	| outgoing-byte-rate-for-broker-<broker-id>    | meter      | Bytes/second written off a given broker                       |
-	| request-rate                                 | meter      | Requests/second sent to all brokers                           |
-	| request-rate-for-broker-<broker-id>          | meter      | Requests/second sent to a given broker                        |
-	| request-size                                 | histogram  | Distribution of the request size in bytes for all brokers     |
-	| request-size-for-broker-<broker-id>          | histogram  | Distribution of the request size in bytes for a given broker  |
-	| request-latency-in-ms                        | histogram  | Distribution of the request latency in ms for all brokers     |
-	| request-latency-in-ms-for-broker-<broker-id> | histogram  | Distribution of the request latency in ms for a given broker  |
-	| response-rate                                | meter      | Responses/second received from all brokers                    |
-	| response-rate-for-broker-<broker-id>         | meter      | Responses/second received from a given broker                 |
-	| response-size                                | histogram  | Distribution of the response size in bytes for all brokers    |
-	| response-size-for-broker-<broker-id>         | histogram  | Distribution of the response size in bytes for a given broker |
-	| requests-in-flight                           | counter    | The current number of in-flight requests awaiting a response  |
-	|                                              |            | for all brokers                                               |
-	| requests-in-flight-for-broker-<broker-id>    | counter    | The current number of in-flight requests awaiting a response  |
-	|                                              |            | for a given broker                                            |
-	+----------------------------------------------+------------+---------------------------------------------------------------+
+	+---------------------------------------------------------+------------+---------------------------------------------------------------+
+	| Name                                                    | Type       | Description                                                   |
+	+---------------------------------------------------------+------------+---------------------------------------------------------------+
+	| incoming-byte-rate                                      | meter      | Bytes/second read off all brokers                             |
+	| incoming-byte-rate-for-broker-<broker-id>               | meter      | Bytes/second read off a given broker                          |
+	| outgoing-byte-rate                                      | meter      | Bytes/second written off all brokers                          |
+	| outgoing-byte-rate-for-broker-<broker-id>               | meter      | Bytes/second written off a given broker                       |
+	| request-rate                                            | meter      | Requests/second sent to all brokers                           |
+	| request-rate-for-broker-<broker-id>                     | meter      | Requests/second sent to a given broker                        |
+	| request-size                                            | histogram  | Distribution of the request size in bytes for all brokers     |
+	| request-size-for-broker-<broker-id>                     | histogram  | Distribution of the request size in bytes for a given broker  |
+	| request-latency-in-ms                                   | histogram  | Distribution of the request latency in ms for all brokers     |
+	| request-latency-in-ms-for-broker-<broker-id>            | histogram  | Distribution of the request latency in ms for a given broker  |
+	| response-rate                                           | meter      | Responses/second received from all brokers                    |
+	| response-rate-for-broker-<broker-id>                    | meter      | Responses/second received from a given broker                 |
+	| response-size                                           | histogram  | Distribution of the response size in bytes for all brokers    |
+	| response-size-for-broker-<broker-id>                    | histogram  | Distribution of the response size in bytes for a given broker |
+	| requests-in-flight                                      | counter    | The current number of in-flight requests awaiting a response  |
+	|                                                         |            | for all brokers                                               |
+	| requests-in-flight-for-broker-<broker-id>               | counter    | The current number of in-flight requests awaiting a response  |
+	|                                                         |            | for a given broker                                            |
+	| protocol-requests-rate-<api-key>          	          | meter      | Number of api requests sent to the brokers for all brokers    |
+	|                                                         |            | https://kafka.apache.org/protocol.html#protocol_api_keys      |                                        |
+	| protocol-requests-rate-<api-key>-for-broker-<broker-id> | meter      | Number of packets sent to the brokers by api-key for a given  |
+	|                                                         |            | broker                                                        |
+	+---------------------------------------------------------+------------+---------------------------------------------------------------+
 
 Note that we do not gather specific metrics for seed brokers but they are part of the "all brokers" metrics.
 
diff --git a/utils.go b/utils.go
index 74c3089..819b659 100644
--- a/utils.go
+++ b/utils.go
@@ -191,6 +191,8 @@ var (
 	V3_2_1_0  = newKafkaVersion(3, 2, 1, 0)
 	V3_2_2_0  = newKafkaVersion(3, 2, 2, 0)
 	V3_2_3_0  = newKafkaVersion(3, 2, 3, 0)
+	V3_3_0_0  = newKafkaVersion(3, 3, 0, 0)
+	V3_3_1_0  = newKafkaVersion(3, 3, 1, 0)
 
 	SupportedVersions = []KafkaVersion{
 		V0_8_2_0,
@@ -244,9 +246,11 @@ var (
 		V3_2_1_0,
 		V3_2_2_0,
 		V3_2_3_0,
+		V3_3_0_0,
+		V3_3_1_0,
 	}
 	MinVersion     = V0_8_2_0
-	MaxVersion     = V3_2_3_0
+	MaxVersion     = V3_3_1_0
 	DefaultVersion = V1_0_0_0
 
 	// reduced set of versions to matrix test
@@ -256,17 +260,13 @@ var (
 		V1_0_2_0,
 		V1_1_1_0,
 		V2_0_1_0,
-		V2_1_1_0,
 		V2_2_2_0,
-		V2_3_1_0,
 		V2_4_1_0,
-		V2_5_1_0,
 		V2_6_2_0,
-		V2_7_1_0,
 		V2_8_2_0,
-		V3_0_2_0,
 		V3_1_2_0,
 		V3_2_3_0,
+		V3_3_1_0,
 	}
 )
 
diff --git a/zstd.go b/zstd.go
index 80507e1..6073ce7 100644
--- a/zstd.go
+++ b/zstd.go
@@ -6,28 +6,49 @@ import (
 	"github.com/klauspost/compress/zstd"
 )
 
+// zstdMaxBufferedEncoders maximum number of not-in-use zstd encoders
+// If the pool of encoders is exhausted then new encoders will be created on the fly
+const zstdMaxBufferedEncoders = 1
+
 type ZstdEncoderParams struct {
 	Level int
 }
 type ZstdDecoderParams struct {
 }
 
-var zstdEncMap, zstdDecMap sync.Map
+var zstdDecMap sync.Map
+
+var zstdAvailableEncoders sync.Map
 
-func getEncoder(params ZstdEncoderParams) *zstd.Encoder {
-	if ret, ok := zstdEncMap.Load(params); ok {
-		return ret.(*zstd.Encoder)
+func getZstdEncoderChannel(params ZstdEncoderParams) chan *zstd.Encoder {
+	if c, ok := zstdAvailableEncoders.Load(params); ok {
+		return c.(chan *zstd.Encoder)
 	}
-	// It's possible to race and create multiple new writers.
-	// Only one will survive GC after use.
-	encoderLevel := zstd.SpeedDefault
-	if params.Level != CompressionLevelDefault {
-		encoderLevel = zstd.EncoderLevelFromZstd(params.Level)
+	c, _ := zstdAvailableEncoders.LoadOrStore(params, make(chan *zstd.Encoder, zstdMaxBufferedEncoders))
+	return c.(chan *zstd.Encoder)
+}
+
+func getZstdEncoder(params ZstdEncoderParams) *zstd.Encoder {
+	select {
+	case enc := <-getZstdEncoderChannel(params):
+		return enc
+	default:
+		encoderLevel := zstd.SpeedDefault
+		if params.Level != CompressionLevelDefault {
+			encoderLevel = zstd.EncoderLevelFromZstd(params.Level)
+		}
+		zstdEnc, _ := zstd.NewWriter(nil, zstd.WithZeroFrames(true),
+			zstd.WithEncoderLevel(encoderLevel),
+			zstd.WithEncoderConcurrency(1))
+		return zstdEnc
+	}
+}
+
+func releaseEncoder(params ZstdEncoderParams, enc *zstd.Encoder) {
+	select {
+	case getZstdEncoderChannel(params) <- enc:
+	default:
 	}
-	zstdEnc, _ := zstd.NewWriter(nil, zstd.WithZeroFrames(true),
-		zstd.WithEncoderLevel(encoderLevel))
-	zstdEncMap.Store(params, zstdEnc)
-	return zstdEnc
 }
 
 func getDecoder(params ZstdDecoderParams) *zstd.Decoder {
@@ -36,7 +57,7 @@ func getDecoder(params ZstdDecoderParams) *zstd.Decoder {
 	}
 	// It's possible to race and create multiple new readers.
 	// Only one will survive GC after use.
-	zstdDec, _ := zstd.NewReader(nil)
+	zstdDec, _ := zstd.NewReader(nil, zstd.WithDecoderConcurrency(0))
 	zstdDecMap.Store(params, zstdDec)
 	return zstdDec
 }
@@ -46,5 +67,8 @@ func zstdDecompress(params ZstdDecoderParams, dst, src []byte) ([]byte, error) {
 }
 
 func zstdCompress(params ZstdEncoderParams, dst, src []byte) ([]byte, error) {
-	return getEncoder(params).EncodeAll(src, dst), nil
+	enc := getZstdEncoder(params)
+	out := enc.EncodeAll(src, dst)
+	releaseEncoder(params, enc)
+	return out, nil
 }


Test patch:
diff --git a/async_producer_test.go b/async_producer_test.go
index 41fb88f..fb895e8 100644
--- a/async_producer_test.go
+++ b/async_producer_test.go
@@ -1302,6 +1302,75 @@ func TestAsyncProducerIdempotentRetryCheckBatch(t *testing.T) {
 	}
 }
 
+// test case for https://github.com/Shopify/sarama/pull/2378
+func TestAsyncProducerIdempotentRetryCheckBatch_2378(t *testing.T) {
+	broker := NewMockBroker(t, 1)
+
+	metadataResponse := &MetadataResponse{
+		Version:      1,
+		ControllerID: 1,
+	}
+	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
+	metadataResponse.AddTopicPartition("my_topic", 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
+
+	initProducerIDResponse := &InitProducerIDResponse{
+		ThrottleTime:  0,
+		ProducerID:    1000,
+		ProducerEpoch: 1,
+	}
+
+	prodNotLeaderResponse := &ProduceResponse{
+		Version:      3,
+		ThrottleTime: 0,
+	}
+	prodNotLeaderResponse.AddTopicPartition("my_topic", 0, ErrNotEnoughReplicas)
+
+	handlerFailBeforeWrite := func(req *request) (res encoderWithHeader) {
+		switch req.body.key() {
+		case 3:
+			return metadataResponse
+		case 22:
+			return initProducerIDResponse
+		case 0: // for msg, always return error to trigger retryBatch
+			return prodNotLeaderResponse
+		}
+		return nil
+	}
+
+	config := NewTestConfig()
+	config.Version = V0_11_0_0
+	config.Producer.Idempotent = true
+	config.Net.MaxOpenRequests = 1
+	config.Producer.Retry.Max = 1 // set max retry to 1
+	config.Producer.RequiredAcks = WaitForAll
+	config.Producer.Return.Successes = true
+	config.Producer.Flush.Frequency = 50 * time.Millisecond
+	config.Producer.Retry.Backoff = 100 * time.Millisecond
+
+	broker.setHandler(handlerFailBeforeWrite)
+	producer, err := NewAsyncProducer([]string{broker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 3; i++ {
+		producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
+	}
+
+	go func() {
+		for i := 0; i < 7; i++ {
+			producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder("goroutine")}
+			time.Sleep(100 * time.Millisecond)
+		}
+	}()
+
+	// this will block until 5 minutes timeout before pr 2378 merge
+	expectResults(t, producer, 0, 10)
+
+	broker.Close()
+	closeProducer(t, producer)
+}
+
 func TestAsyncProducerIdempotentErrorOnOutOfSeq(t *testing.T) {
 	broker := NewMockBroker(t, 1)
 
@@ -1711,7 +1780,7 @@ func TestTxnProduceBumpEpoch(t *testing.T) {
 	config.ApiVersionsRequest = false
 
 	metadataLeader := new(MetadataResponse)
-	metadataLeader.Version = 5
+	metadataLeader.Version = 7
 	metadataLeader.ControllerID = broker.brokerID
 	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
 	metadataLeader.AddTopic("test-topic", ErrNoError)
diff --git a/fetch_request_test.go b/fetch_request_test.go
index 0b807b9..3df2df1 100644
--- a/fetch_request_test.go
+++ b/fetch_request_test.go
@@ -41,7 +41,7 @@ var (
 		0x00, 0x05, 't', 'o', 'p', 'i', 'c',
 		0x00, 0x00, 0x00, 0x01,
 		0x00, 0x00, 0x00, 0x12, // partitionID
-		0xFF, 0xFF, 0xFF, 0xFF, // currentLeaderEpoch
+		0x00, 0x00, 0x00, 0x66, // currentLeaderEpoch
 		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x34, // fetchOffset
 		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // logStartOffset
 		0x00, 0x00, 0x00, 0x56, // maxBytes
@@ -67,7 +67,7 @@ func TestFetchRequest(t *testing.T) {
 		request := new(FetchRequest)
 		request.MaxWaitTime = 0
 		request.MinBytes = 0
-		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		request.AddBlock("topic", 0x12, 0x34, 0x56, -1)
 		testRequest(t, "one block", request, fetchRequestOneBlock)
 	})
 
@@ -76,18 +76,18 @@ func TestFetchRequest(t *testing.T) {
 		request.Version = 4
 		request.MaxBytes = 0xFF
 		request.Isolation = ReadCommitted
-		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		request.AddBlock("topic", 0x12, 0x34, 0x56, -1)
 		testRequest(t, "one block v4", request, fetchRequestOneBlockV4)
 	})
 
-	t.Run("one block v11 rackid", func(t *testing.T) {
+	t.Run("one block v11 rackid and leader epoch", func(t *testing.T) {
 		request := new(FetchRequest)
 		request.Version = 11
 		request.MaxBytes = 0xFF
 		request.Isolation = ReadCommitted
 		request.SessionID = 0xAA
 		request.SessionEpoch = 0xEE
-		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		request.AddBlock("topic", 0x12, 0x34, 0x56, 0x66)
 		request.RackID = "rack01"
 		testRequest(t, "one block v11 rackid", request, fetchRequestOneBlockV11)
 	})
diff --git a/zstd_test.go b/zstd_test.go
new file mode 100644
index 0000000..efdc6d8
--- /dev/null
+++ b/zstd_test.go
@@ -0,0 +1,29 @@
+package sarama
+
+import (
+	"runtime"
+	"testing"
+)
+
+func BenchmarkZstdMemoryConsumption(b *testing.B) {
+	params := ZstdEncoderParams{Level: 9}
+	buf := make([]byte, 1024*1024)
+	for i := 0; i < len(buf); i++ {
+		buf[i] = byte((i / 256) + (i * 257))
+	}
+
+	cpus := 96
+
+	gomaxprocsBackup := runtime.GOMAXPROCS(cpus)
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 2*cpus; j++ {
+			_, _ = zstdCompress(params, nil, buf)
+		}
+		// drain the buffered encoder
+		getZstdEncoder(params)
+		// previously this would be achieved with
+		// zstdEncMap.Delete(params)
+	}
+	runtime.GOMAXPROCS(gomaxprocsBackup)
+}