Repository: Shopify/sarama
Pull request #1560 (2019-12-16 13:59:27)

add sync pool for gzip 1-9
when large number of messages, for gzip(1-9) cost high cpu, because of gc.


Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
new file mode 100644
index 0000000..5621bb3
--- /dev/null
+++ b/.github/CODEOWNERS
@@ -0,0 +1,1 @@
+* @bai
diff --git a/.github/probots.yml b/.github/probots.yml
index 1491d27..b6fae4b 100644
--- a/.github/probots.yml
+++ b/.github/probots.yml
@@ -1,2 +1,3 @@
 enabled:
-  - cla
+- cla
+- probot-stale
diff --git a/.github/stale.yml b/.github/stale.yml
new file mode 100644
index 0000000..e597ccb
--- /dev/null
+++ b/.github/stale.yml
@@ -0,0 +1,34 @@
+# Number of days of inactivity before an Issue or Pull Request becomes stale.
+daysUntilStale: 90
+
+# Number of days of inactivity before an Issue or Pull Request with the stale label is closed.
+# Set to false to disable. If disabled, issues still need to be closed manually, but will remain marked as stale.
+daysUntilClose: 30
+
+# Issues or Pull Requests with these labels will never be considered stale. Set to `[]` to disable.
+exemptLabels: ["stale/exempt", "pinned"]
+
+# Label to use when marking as stale.
+staleLabel: stale
+
+# Comment to post when marking as stale. Set to `false` to disable.
+issues:
+  markComment: >
+    Thank you for taking the time to raise this issue. However, it has
+    not had any activity on it in the past 90 days and will be closed in
+    30 days if no updates occur.
+
+    Please check if the master branch has already resolved the issue
+    since it was raised. If you believe the issue is still valid and you
+    would like input from the maintainers then please comment to ask for
+    it to be reviewed.
+
+pulls:
+  markComment: >
+    Thank you for your contribution! However, this pull request has not
+    had any activity in the past 90 days and will be closed in 30 days if
+    no updates occur.
+
+    If you believe the changes are still valid then please verify your
+    branch has no conflicts with master and rebase if needed. If you
+    are awaiting a (re-)review then please let us know.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..a4a14c6
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,63 @@
+name: CI
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    name: Go ${{ matrix.go-version }} with Kafka ${{ matrix.kafka-version }} on Ubuntu
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        go-version: [1.12.x, 1.13.x]
+        kafka-version: [2.2.2, 2.3.1, 2.4.0]
+        platform: [ubuntu-latest]
+
+    env:
+      KAFKA_PEERS: localhost:9091,localhost:9092,localhost:9093,localhost:9094,localhost:9095
+      TOXIPROXY_ADDR: http://localhost:8474
+      KAFKA_INSTALL_ROOT: /home/runner/kafka
+      KAFKA_HOSTNAME: localhost
+      DEBUG: true
+      KAFKA_VERSION: ${{ matrix.kafka-version }}
+      KAFKA_SCALA_VERSION: 2.12
+
+    steps:
+    - uses: actions/checkout@v1
+
+    - name: Setup Go
+      uses: actions/setup-go@v1
+      with:
+        go-version: ${{ matrix.go-version }}
+
+    - uses: actions/cache@v1
+      with:
+        path: ~/go/pkg/mod
+        key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+        restore-keys: |
+          ${{ runner.os }}-go-
+
+    # See https://github.com/actions/setup-go/issues/14
+    - name: Setup env
+      run: |
+        echo "::set-env name=GOPATH::$(go env GOPATH)"
+        echo "::add-path::$(go env GOPATH)/bin"
+      shell: bash
+
+    - name: Install dependencies
+      run: |
+        curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.23.3
+        export REPOSITORY_ROOT=${GITHUB_WORKSPACE}
+        vagrant/install_cluster.sh
+        vagrant/boot_cluster.sh
+        vagrant/create_topics.sh
+        vagrant/run_java_producer.sh
+
+    - name: Run test suite
+      run: make test
+
+    - name: Run linter
+      run: make lint
+
+    - name: Teardown
+      run: vagrant/halt_cluster.sh
diff --git a/.golangci.yml b/.golangci.yml
new file mode 100644
index 0000000..47624f3
--- /dev/null
+++ b/.golangci.yml
@@ -0,0 +1,74 @@
+run:
+  timeout: 5m
+  deadline: 10m
+
+linters-settings:
+  govet:
+    check-shadowing: false
+  golint:
+    min-confidence: 0
+  gocyclo:
+    min-complexity: 99
+  maligned:
+    suggest-new: true
+  dupl:
+    threshold: 100
+  goconst:
+    min-len: 2
+    min-occurrences: 3
+  misspell:
+    locale: US
+  goimports:
+    local-prefixes: github.com/Shopify/sarama
+  gocritic:
+    enabled-tags:
+      - diagnostic
+      - experimental
+      - opinionated
+      - performance
+      - style
+    disabled-checks:
+      - wrapperFunc
+      - ifElseChain
+  funlen:
+    lines: 300
+    statements: 300
+
+linters:
+  disable-all: true
+  enable:
+    - bodyclose
+    - deadcode
+    - depguard
+    - dogsled
+    # - dupl
+    - errcheck
+    - funlen
+    # - gocritic
+    - gocyclo
+    - gofmt
+    - goimports
+    # - golint
+    - gosec
+    # - gosimple
+    - govet
+    # - ineffassign
+    - interfacer
+    # - misspell
+    # - nakedret
+    # - scopelint
+    # - staticcheck
+    - structcheck
+    # - stylecheck
+    - typecheck
+    - unconvert
+    - unused
+    - varcheck
+    - whitespace
+    # - goconst
+    # - gochecknoinits
+
+issues:
+  exclude:
+    - consider giving a name to these results
+    - include an explanation for nolint directive
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index cace313..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,37 +0,0 @@
-dist: xenial
-language: go
-go:
-- 1.12.x
-- 1.13.x
-
-env:
-  global:
-  - KAFKA_PEERS=localhost:9091,localhost:9092,localhost:9093,localhost:9094,localhost:9095
-  - TOXIPROXY_ADDR=http://localhost:8474
-  - KAFKA_INSTALL_ROOT=/home/travis/kafka
-  - KAFKA_HOSTNAME=localhost
-  - DEBUG=true
-  matrix:
-  - KAFKA_VERSION=2.2.1 KAFKA_SCALA_VERSION=2.12
-  - KAFKA_VERSION=2.3.0 KAFKA_SCALA_VERSION=2.12
-
-before_install:
-- export REPOSITORY_ROOT=${TRAVIS_BUILD_DIR}
-- vagrant/install_cluster.sh
-- vagrant/boot_cluster.sh
-- vagrant/create_topics.sh
-- vagrant/run_java_producer.sh
-
-install: make install_dependencies
-
-script:
-- make test
-- make vet
-- make errcheck
-- if [[ "$TRAVIS_GO_VERSION" == 1.13* ]]; then make fmt; fi
-
-after_success:
-- go tool cover -func coverage.txt
-- bash <(curl -s https://codecov.io/bash)
-
-after_script: vagrant/halt_cluster.sh
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dfa7e75..2bebeb1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,74 @@
 # Changelog
 
+#### Unreleased
+
+#### Version 1.26.1 (2020-02-04)
+
+Improvements:
+- Add requests-in-flight metric ([1539](https://github.com/Shopify/sarama/pull/1539))
+- Fix misleading example for cluster admin ([1595](https://github.com/Shopify/sarama/pull/1595))
+- Replace Travis with GitHub Actions, linters housekeeping ([1573](https://github.com/Shopify/sarama/pull/1573))
+- Allow BalanceStrategy to provide custom assignment data ([1592](https://github.com/Shopify/sarama/pull/1592))
+
+Bug Fixes:
+- Adds back Consumer.Offsets.CommitInterval to fix API ([1590](https://github.com/Shopify/sarama/pull/1590))
+- Fix error message s/CommitInterval/AutoCommit.Interval ([1589](https://github.com/Shopify/sarama/pull/1589))
+
+#### Version 1.26.0 (2020-01-24)
+
+New Features:
+- Enable zstd compression
+  ([1574](https://github.com/Shopify/sarama/pull/1574),
+  [1582](https://github.com/Shopify/sarama/pull/1582))
+- Support headers in tools kafka-console-producer
+  ([1549](https://github.com/Shopify/sarama/pull/1549))
+
+Improvements:
+- Add SASL AuthIdentity to SASL frames (authzid)
+  ([1585](https://github.com/Shopify/sarama/pull/1585)).
+
+Bug Fixes:
+- Sending messages with ZStd compression enabled fails in multiple ways
+  ([1252](https://github.com/Shopify/sarama/issues/1252)).
+- Use the broker for any admin on BrokerConfig
+  ([1571](https://github.com/Shopify/sarama/pull/1571)).
+- Set DescribeConfigRequest Version field
+  ([1576](https://github.com/Shopify/sarama/pull/1576)).
+- ConsumerGroup flooding logs with client/metadata update req
+  ([1578](https://github.com/Shopify/sarama/pull/1578)).
+- MetadataRequest version in DescribeCluster
+  ([1580](https://github.com/Shopify/sarama/pull/1580)).
+- Fix deadlock in consumer group handleError
+  ([1581](https://github.com/Shopify/sarama/pull/1581))
+- Fill in the Fetch{Request,Response} protocol
+  ([1582](https://github.com/Shopify/sarama/pull/1582)).
+- Retry topic request on ControllerNotAvailable
+  ([1586](https://github.com/Shopify/sarama/pull/1586)).
+
+#### Version 1.25.0 (2020-01-13)
+
+New Features:
+- Support TLS protocol in kafka-producer-performance
+  ([1538](https://github.com/Shopify/sarama/pull/1538)).
+- Add support for kafka 2.4.0
+  ([1552](https://github.com/Shopify/sarama/pull/1552)).
+
+Improvements:
+- Allow the Consumer to disable auto-commit offsets
+  ([1164](https://github.com/Shopify/sarama/pull/1164)).
+- Produce records with consistent timestamps
+  ([1455](https://github.com/Shopify/sarama/pull/1455)).
+
+Bug Fixes:
+- Fix incorrect SetTopicMetadata name mentions
+  ([1534](https://github.com/Shopify/sarama/pull/1534)).
+- Fix client.tryRefreshMetadata Println
+  ([1535](https://github.com/Shopify/sarama/pull/1535)).
+- Fix panic on calling updateMetadata on closed client
+  ([1531](https://github.com/Shopify/sarama/pull/1531)).
+- Fix possible faulty metrics in TestFuncProducing
+  ([1545](https://github.com/Shopify/sarama/pull/1545)).
+
 #### Version 1.24.1 (2019-10-31)
 
 New Features:
diff --git a/Makefile b/Makefile
index 9c8329e..c3b431a 100644
--- a/Makefile
+++ b/Makefile
@@ -1,56 +1,27 @@
-export GO111MODULE=on
+default: fmt get update test lint
 
-default: fmt vet errcheck test lint
+GO       := GO111MODULE=on GOPRIVATE=github.com/linkedin GOSUMDB=off go
+GOBUILD  := CGO_ENABLED=0 $(GO) build $(BUILD_FLAG)
+GOTEST   := $(GO) test -gcflags='-l' -p 3 -v -race -timeout 6m -coverprofile=profile.out -covermode=atomic
 
-# Taken from https://github.com/codecov/example-go#caveat-multiple-files
-.PHONY: test
-test:
-	echo "mode: atomic" > coverage.txt
-	for d in `go list ./...`; do \
-		go test -p 1 -v -timeout 6m -race -coverprofile=profile.out -covermode=atomic $$d || exit 1; \
-		if [ -f profile.out ]; then \
-			tail +2 profile.out >> coverage.txt; \
-			rm profile.out; \
-		fi \
-	done
-
-GOLINT := $(shell command -v golint)
+FILES    := $(shell find . -name '*.go' -type f -not -name '*.pb.go' -not -name '*_generated.go' -not -name '*_test.go')
+TESTS    := $(shell find . -name '*.go' -type f -not -name '*.pb.go' -not -name '*_generated.go' -name '*_test.go')
 
-.PHONY: lint
-lint:
-ifndef GOLINT
-	go get golang.org/x/lint/golint
-endif
-	go list ./... | xargs golint
-
-.PHONY: vet
-vet:
-	go vet ./...
+get:
+	$(GO) get ./...
+	$(GO) mod verify
+	$(GO) mod tidy
 
-ERRCHECK := $(shell command -v errcheck)
-# See https://github.com/kisielk/errcheck/pull/141 for details on ignorepkg
-.PHONY: errcheck
-errcheck:
-ifndef ERRCHECK
-	go get github.com/kisielk/errcheck
-endif
-	errcheck -ignorepkg fmt github.com/Shopify/sarama/...
+update:
+	$(GO) get -u -v all
+	$(GO) mod verify
+	$(GO) mod tidy
 
-.PHONY: fmt
 fmt:
-	@if [ -n "$$(go fmt ./...)" ]; then echo 'Please run go fmt on your code.' && exit 1; fi
+	gofmt -s -l -w $(FILES) $(TESTS)
 
-.PHONY : install_dependencies
-install_dependencies: get
-
-.PHONY: get
-get:
-	go get -v ./...
-
-.PHONY: clean
-clean:
-	go clean ./...
+lint:
+	golangci-lint run
 
-.PHONY: tidy
-tidy:
-	go mod tidy -v
+test:
+	$(GOTEST) ./...
diff --git a/Vagrantfile b/Vagrantfile
index f4b848a..07d7ffb 100644
--- a/Vagrantfile
+++ b/Vagrantfile
@@ -1,14 +1,8 @@
-# -*- mode: ruby -*-
-# vi: set ft=ruby :
-
-# Vagrantfile API/syntax version. Don't touch unless you know what you're doing!
-VAGRANTFILE_API_VERSION = "2"
-
 # We have 5 * 192MB ZK processes and 5 * 320MB Kafka processes => 2560MB
 MEMORY = 3072
 
-Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
-  config.vm.box = "ubuntu/trusty64"
+Vagrant.configure("2") do |config|
+  config.vm.box = "ubuntu/bionic64"
 
   config.vm.provision :shell, path: "vagrant/provision.sh"
 
diff --git a/acl_delete_response.go b/acl_delete_response.go
index 6529565..3754fae 100644
--- a/acl_delete_response.go
+++ b/acl_delete_response.go
@@ -53,7 +53,7 @@ func (d *DeleteAclsResponse) key() int16 {
 }
 
 func (d *DeleteAclsResponse) version() int16 {
-	return int16(d.Version)
+	return d.Version
 }
 
 func (d *DeleteAclsResponse) requiredVersion() KafkaVersion {
diff --git a/acl_describe_response.go b/acl_describe_response.go
index 12126e5..a60d39f 100644
--- a/acl_describe_response.go
+++ b/acl_describe_response.go
@@ -74,7 +74,7 @@ func (d *DescribeAclsResponse) key() int16 {
 }
 
 func (d *DescribeAclsResponse) version() int16 {
-	return int16(d.Version)
+	return d.Version
 }
 
 func (d *DescribeAclsResponse) requiredVersion() KafkaVersion {
diff --git a/admin.go b/admin.go
index 6c9b1e9..7dd7250 100644
--- a/admin.go
+++ b/admin.go
@@ -2,8 +2,11 @@ package sarama
 
 import (
 	"errors"
+	"fmt"
 	"math/rand"
+	"strconv"
 	"sync"
+	"time"
 )
 
 // ClusterAdmin is the administrative client for Kafka, which supports managing and inspecting topics,
@@ -132,8 +135,45 @@ func (ca *clusterAdmin) Controller() (*Broker, error) {
 	return ca.client.Controller()
 }
 
-func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateOnly bool) error {
+func (ca *clusterAdmin) refreshController() (*Broker, error) {
+	return ca.client.RefreshController()
+}
+
+// isErrNoController returns `true` if the given error type unwraps to an
+// `ErrNotController` response from Kafka
+func isErrNoController(err error) bool {
+	switch e := err.(type) {
+	case *TopicError:
+		return e.Err == ErrNotController
+	case *TopicPartitionError:
+		return e.Err == ErrNotController
+	case KError:
+		return e == ErrNotController
+	}
+	return false
+}
 
+// retryOnError will repeatedly call the given (error-returning) func in the
+// case that its response is non-nil and retriable (as determined by the
+// provided retriable func) up to the maximum number of tries permitted by
+// the admin client configuration
+func (ca *clusterAdmin) retryOnError(retriable func(error) bool, fn func() error) error {
+	var err error
+	for attempt := 0; attempt < ca.conf.Admin.Retry.Max; attempt++ {
+		err = fn()
+		if err == nil || !retriable(err) {
+			return err
+		}
+		Logger.Printf(
+			"admin/request retrying after %dms... (%d attempts remaining)\n",
+			ca.conf.Admin.Retry.Backoff/time.Millisecond, ca.conf.Admin.Retry.Max-attempt)
+		time.Sleep(ca.conf.Admin.Retry.Backoff)
+		continue
+	}
+	return err
+}
+
+func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateOnly bool) error {
 	if topic == "" {
 		return ErrInvalidTopic
 	}
@@ -158,26 +198,31 @@ func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateO
 		request.Version = 2
 	}
 
-	b, err := ca.Controller()
-	if err != nil {
-		return err
-	}
+	return ca.retryOnError(isErrNoController, func() error {
+		b, err := ca.Controller()
+		if err != nil {
+			return err
+		}
 
-	rsp, err := b.CreateTopics(request)
-	if err != nil {
-		return err
-	}
+		rsp, err := b.CreateTopics(request)
+		if err != nil {
+			return err
+		}
 
-	topicErr, ok := rsp.TopicErrors[topic]
-	if !ok {
-		return ErrIncompleteResponse
-	}
+		topicErr, ok := rsp.TopicErrors[topic]
+		if !ok {
+			return ErrIncompleteResponse
+		}
 
-	if topicErr.Err != ErrNoError {
-		return topicErr
-	}
+		if topicErr.Err != ErrNoError {
+			if topicErr.Err == ErrNotController {
+				_, _ = ca.refreshController()
+			}
+			return topicErr
+		}
 
-	return nil
+		return nil
+	})
 }
 
 func (ca *clusterAdmin) DescribeTopics(topics []string) (metadata []*TopicMetadata, err error) {
@@ -214,7 +259,7 @@ func (ca *clusterAdmin) DescribeCluster() (brokers []*Broker, controllerID int32
 		Topics: []string{},
 	}
 
-	if ca.conf.Version.IsAtLeast(V0_11_0_0) {
+	if ca.conf.Version.IsAtLeast(V0_10_0_0) {
 		request.Version = 1
 	}
 
@@ -226,6 +271,16 @@ func (ca *clusterAdmin) DescribeCluster() (brokers []*Broker, controllerID int32
 	return response.Brokers, response.ControllerID, nil
 }
 
+func (ca *clusterAdmin) findBroker(id int32) (*Broker, error) {
+	brokers := ca.client.Brokers()
+	for _, b := range brokers {
+		if b.ID() == id {
+			return b, nil
+		}
+	}
+	return nil, fmt.Errorf("could not find broker id %d", id)
+}
+
 func (ca *clusterAdmin) findAnyBroker() (*Broker, error) {
 	brokers := ca.client.Brokers()
 	if len(brokers) > 0 {
@@ -283,6 +338,15 @@ func (ca *clusterAdmin) ListTopics() (map[string]TopicDetail, error) {
 	describeConfigsReq := &DescribeConfigsRequest{
 		Resources: describeConfigsResources,
 	}
+
+	if ca.conf.Version.IsAtLeast(V1_1_0_0) {
+		describeConfigsReq.Version = 1
+	}
+
+	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+		describeConfigsReq.Version = 2
+	}
+
 	describeConfigsResp, err := b.DescribeConfigs(describeConfigsReq)
 	if err != nil {
 		return nil, err
@@ -308,7 +372,6 @@ func (ca *clusterAdmin) ListTopics() (map[string]TopicDetail, error) {
 }
 
 func (ca *clusterAdmin) DeleteTopic(topic string) error {
-
 	if topic == "" {
 		return ErrInvalidTopic
 	}
@@ -322,25 +385,31 @@ func (ca *clusterAdmin) DeleteTopic(topic string) error {
 		request.Version = 1
 	}
 
-	b, err := ca.Controller()
-	if err != nil {
-		return err
-	}
+	return ca.retryOnError(isErrNoController, func() error {
+		b, err := ca.Controller()
+		if err != nil {
+			return err
+		}
 
-	rsp, err := b.DeleteTopics(request)
-	if err != nil {
-		return err
-	}
+		rsp, err := b.DeleteTopics(request)
+		if err != nil {
+			return err
+		}
 
-	topicErr, ok := rsp.TopicErrorCodes[topic]
-	if !ok {
-		return ErrIncompleteResponse
-	}
+		topicErr, ok := rsp.TopicErrorCodes[topic]
+		if !ok {
+			return ErrIncompleteResponse
+		}
 
-	if topicErr != ErrNoError {
-		return topicErr
-	}
-	return nil
+		if topicErr != ErrNoError {
+			if topicErr == ErrNotController {
+				_, _ = ca.refreshController()
+			}
+			return topicErr
+		}
+
+		return nil
+	})
 }
 
 func (ca *clusterAdmin) CreatePartitions(topic string, count int32, assignment [][]int32, validateOnly bool) error {
@@ -356,30 +425,34 @@ func (ca *clusterAdmin) CreatePartitions(topic string, count int32, assignment [
 		Timeout:         ca.conf.Admin.Timeout,
 	}
 
-	b, err := ca.Controller()
-	if err != nil {
-		return err
-	}
+	return ca.retryOnError(isErrNoController, func() error {
+		b, err := ca.Controller()
+		if err != nil {
+			return err
+		}
 
-	rsp, err := b.CreatePartitions(request)
-	if err != nil {
-		return err
-	}
+		rsp, err := b.CreatePartitions(request)
+		if err != nil {
+			return err
+		}
 
-	topicErr, ok := rsp.TopicPartitionErrors[topic]
-	if !ok {
-		return ErrIncompleteResponse
-	}
+		topicErr, ok := rsp.TopicPartitionErrors[topic]
+		if !ok {
+			return ErrIncompleteResponse
+		}
 
-	if topicErr.Err != ErrNoError {
-		return topicErr
-	}
+		if topicErr.Err != ErrNoError {
+			if topicErr.Err == ErrNotController {
+				_, _ = ca.refreshController()
+			}
+			return topicErr
+		}
 
-	return nil
+		return nil
+	})
 }
 
 func (ca *clusterAdmin) DeleteRecords(topic string, partitionOffsets map[int32]int64) error {
-
 	if topic == "" {
 		return ErrInvalidTopic
 	}
@@ -432,8 +505,14 @@ func (ca *clusterAdmin) DeleteRecords(topic string, partitionOffsets map[int32]i
 	return nil
 }
 
-func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry, error) {
+// Returns a bool indicating whether the resource request needs to go to a
+// specific broker
+func dependsOnSpecificNode(resource ConfigResource) bool {
+	return (resource.Type == BrokerResource && resource.Name != "") ||
+		resource.Type == BrokerLoggerResource
+}
 
+func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry, error) {
 	var entries []ConfigEntry
 	var resources []*ConfigResource
 	resources = append(resources, &resource)
@@ -442,11 +521,31 @@ func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry,
 		Resources: resources,
 	}
 
-	b, err := ca.Controller()
+	if ca.conf.Version.IsAtLeast(V1_1_0_0) {
+		request.Version = 1
+	}
+
+	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+		request.Version = 2
+	}
+
+	var (
+		b   *Broker
+		err error
+	)
+
+	// DescribeConfig of broker/broker logger must be sent to the broker in question
+	if dependsOnSpecificNode(resource) {
+		id, _ := strconv.Atoi(resource.Name)
+		b, err = ca.findBroker(int32(id))
+	} else {
+		b, err = ca.findAnyBroker()
+	}
 	if err != nil {
 		return nil, err
 	}
 
+	_ = b.Open(ca.client.Config())
 	rsp, err := b.DescribeConfigs(request)
 	if err != nil {
 		return nil, err
@@ -466,7 +565,6 @@ func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry,
 }
 
 func (ca *clusterAdmin) AlterConfig(resourceType ConfigResourceType, name string, entries map[string]*string, validateOnly bool) error {
-
 	var resources []*AlterConfigsResource
 	resources = append(resources, &AlterConfigsResource{
 		Type:          resourceType,
@@ -479,11 +577,23 @@ func (ca *clusterAdmin) AlterConfig(resourceType ConfigResourceType, name string
 		ValidateOnly: validateOnly,
 	}
 
-	b, err := ca.Controller()
+	var (
+		b   *Broker
+		err error
+	)
+
+	// AlterConfig of broker/broker logger must be sent to the broker in question
+	if dependsOnSpecificNode(ConfigResource{Name: name, Type: resourceType}) {
+		id, _ := strconv.Atoi(name)
+		b, err = ca.findBroker(int32(id))
+	} else {
+		b, err = ca.findAnyBroker()
+	}
 	if err != nil {
 		return err
 	}
 
+	_ = b.Open(ca.client.Config())
 	rsp, err := b.AlterConfigs(request)
 	if err != nil {
 		return err
@@ -518,7 +628,6 @@ func (ca *clusterAdmin) CreateACL(resource Resource, acl Acl) error {
 }
 
 func (ca *clusterAdmin) ListAcls(filter AclFilter) ([]ResourceAcls, error) {
-
 	request := &DescribeAclsRequest{AclFilter: filter}
 
 	if ca.conf.Version.IsAtLeast(V2_0_0_0) {
@@ -566,7 +675,6 @@ func (ca *clusterAdmin) DeleteACL(filter AclFilter, validateOnly bool) ([]Matchi
 		for _, mACL := range fr.MatchingAcls {
 			mAcls = append(mAcls, *mACL)
 		}
-
 	}
 	return mAcls, nil
 }
@@ -580,7 +688,6 @@ func (ca *clusterAdmin) DescribeConsumerGroups(groups []string) (result []*Group
 			return nil, err
 		}
 		groupsPerBroker[controller] = append(groupsPerBroker[controller], group)
-
 	}
 
 	for broker, brokerGroups := range groupsPerBroker {
@@ -623,7 +730,6 @@ func (ca *clusterAdmin) ListConsumerGroups() (allGroups map[string]string, err e
 			}
 
 			groupMaps <- groups
-
 		}(b, ca.conf)
 	}
 
diff --git a/async_producer.go b/async_producer.go
index 9b15cd1..5c57a81 100644
--- a/async_producer.go
+++ b/async_producer.go
@@ -411,7 +411,7 @@ func (tp *topicProducer) partitionMessage(msg *ProducerMessage) error {
 	var partitions []int32
 
 	err := tp.breaker.Run(func() (err error) {
-		var requiresConsistency = false
+		requiresConsistency := false
 		if ep, ok := tp.partitioner.(DynamicConsistencyPartitioner); ok {
 			requiresConsistency = ep.MessageRequiresConsistency(msg)
 		} else {
diff --git a/balance_strategy.go b/balance_strategy.go
index 67c4d96..d9789a0 100644
--- a/balance_strategy.go
+++ b/balance_strategy.go
@@ -47,6 +47,10 @@ type BalanceStrategy interface {
 	// Plan accepts a map of `memberID -> metadata` and a map of `topic -> partitions`
 	// and returns a distribution plan.
 	Plan(members map[string]ConsumerGroupMemberMetadata, topics map[string][]int32) (BalanceStrategyPlan, error)
+
+	// AssignmentData returns the serialized assignment data for the specified
+	// memberID
+	AssignmentData(memberID string, topics map[string][]int32, generationID int32) ([]byte, error)
 }
 
 // --------------------------------------------------------------------
@@ -132,6 +136,11 @@ func (s *balanceStrategy) Plan(members map[string]ConsumerGroupMemberMetadata, t
 	return plan, nil
 }
 
+// AssignmentData simple strategies do not require any shared assignment data
+func (s *balanceStrategy) AssignmentData(memberID string, topics map[string][]int32, generationID int32) ([]byte, error) {
+	return nil, nil
+}
+
 type balanceStrategySortable struct {
 	topic     string
 	memberIDs []string
@@ -258,7 +267,7 @@ func (s *stickyBalanceStrategy) Plan(members map[string]ConsumerGroupMemberMetad
 	plan := make(BalanceStrategyPlan, len(currentAssignment))
 	for memberID, assignments := range currentAssignment {
 		if len(assignments) == 0 {
-			plan[memberID] = make(map[string][]int32, 0)
+			plan[memberID] = make(map[string][]int32)
 		} else {
 			for _, assignment := range assignments {
 				plan.Add(memberID, assignment.Topic, assignment.Partition)
@@ -268,6 +277,15 @@ func (s *stickyBalanceStrategy) Plan(members map[string]ConsumerGroupMemberMetad
 	return plan, nil
 }
 
+// AssignmentData serializes the set of topics currently assigned to the
+// specified member as part of the supplied balance plan
+func (s *stickyBalanceStrategy) AssignmentData(memberID string, topics map[string][]int32, generationID int32) ([]byte, error) {
+	return encode(&StickyAssignorUserDataV1{
+		Topics:     topics,
+		Generation: generationID,
+	}, nil)
+}
+
 func strsContains(s []string, value string) bool {
 	for _, entry := range s {
 		if entry == value {
@@ -671,14 +689,6 @@ func sortPartitionsByPotentialConsumerAssignments(partition2AllPotentialConsumer
 	return sortedPartionIDs
 }
 
-func deepCopyPartitions(src []topicPartitionAssignment) []topicPartitionAssignment {
-	dst := make([]topicPartitionAssignment, len(src))
-	for i, partition := range src {
-		dst[i] = partition
-	}
-	return dst
-}
-
 func deepCopyAssignment(assignment map[string][]topicPartitionAssignment) map[string][]topicPartitionAssignment {
 	copy := make(map[string][]topicPartitionAssignment, len(assignment))
 	for memberID, subscriptions := range assignment {
@@ -938,9 +948,7 @@ func (p *partitionMovements) in(cycle []string, cycles [][]string) bool {
 	for i := 0; i < len(cycle)-1; i++ {
 		superCycle[i] = cycle[i]
 	}
-	for _, c := range cycle {
-		superCycle = append(superCycle, c)
-	}
+	superCycle = append(superCycle, cycle...)
 	for _, foundCycle := range cycles {
 		if len(foundCycle) == len(cycle) && indexOfSubList(superCycle, foundCycle) != -1 {
 			return true
diff --git a/broker.go b/broker.go
index 8146749..9ca41c9 100644
--- a/broker.go
+++ b/broker.go
@@ -40,6 +40,7 @@ type Broker struct {
 	outgoingByteRate       metrics.Meter
 	responseRate           metrics.Meter
 	responseSize           metrics.Histogram
+	requestsInFlight       metrics.Counter
 	brokerIncomingByteRate metrics.Meter
 	brokerRequestRate      metrics.Meter
 	brokerRequestSize      metrics.Histogram
@@ -47,6 +48,7 @@ type Broker struct {
 	brokerOutgoingByteRate metrics.Meter
 	brokerResponseRate     metrics.Meter
 	brokerResponseSize     metrics.Histogram
+	brokerRequestsInFlight metrics.Counter
 
 	kerberosAuthenticator GSSAPIKerberosAuth
 }
@@ -182,6 +184,7 @@ func (b *Broker) Open(conf *Config) error {
 		b.outgoingByteRate = metrics.GetOrRegisterMeter("outgoing-byte-rate", conf.MetricRegistry)
 		b.responseRate = metrics.GetOrRegisterMeter("response-rate", conf.MetricRegistry)
 		b.responseSize = getOrRegisterHistogram("response-size", conf.MetricRegistry)
+		b.requestsInFlight = metrics.GetOrRegisterCounter("requests-in-flight", conf.MetricRegistry)
 		// 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 {
@@ -189,7 +192,6 @@ func (b *Broker) Open(conf *Config) error {
 		}
 
 		if conf.Net.SASL.Enable {
-
 			b.connErr = b.authenticateViaSASL()
 
 			if b.connErr != nil {
@@ -713,16 +715,19 @@ func (b *Broker) send(rb protocolBody, promiseResponse bool) (*responsePromise,
 	}
 
 	requestTime := time.Now()
+	// Will be decremented in responseReceiver (except error or request with NoResponse)
+	b.addRequestInFlightMetrics(1)
 	bytes, err := b.write(buf)
 	b.updateOutgoingCommunicationMetrics(bytes)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		return nil, err
 	}
 	b.correlationID++
 
 	if !promiseResponse {
 		// Record request latency without the response
-		b.updateRequestLatencyMetrics(time.Since(requestTime))
+		b.updateRequestLatencyAndInFlightMetrics(time.Since(requestTime))
 		return nil, nil
 	}
 
@@ -817,6 +822,9 @@ func (b *Broker) responseReceiver() {
 
 	for response := range b.responses {
 		if dead != nil {
+			// This was previously incremented in send() and
+			// we are not calling updateIncomingCommunicationMetrics()
+			b.addRequestInFlightMetrics(-1)
 			response.errors <- dead
 			continue
 		}
@@ -892,9 +900,12 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	}
 
 	requestTime := time.Now()
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	bytes, err := b.write(buf)
 	b.updateOutgoingCommunicationMetrics(bytes)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to send SASL handshake %s: %s\n", b.addr, err.Error())
 		return err
 	}
@@ -903,6 +914,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	header := make([]byte, 8) // response header
 	_, err = b.readFull(header)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read SASL handshake header : %s\n", err.Error())
 		return err
 	}
@@ -911,6 +923,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	payload := make([]byte, length-4)
 	n, err := b.readFull(payload)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read SASL handshake payload : %s\n", err.Error())
 		return err
 	}
@@ -961,7 +974,6 @@ func (b *Broker) sendAndReceiveSASLPlainAuth() error {
 	// default to V0 to allow for backward compatability when SASL is enabled
 	// but not the handshake
 	if b.conf.Net.SASL.Handshake {
-
 		handshakeErr := b.sendAndReceiveSASLHandshake(SASLTypePlaintext, b.conf.Net.SASL.Version)
 		if handshakeErr != nil {
 			Logger.Printf("Error while performing SASL handshake %s\n", b.addr)
@@ -977,16 +989,18 @@ func (b *Broker) sendAndReceiveSASLPlainAuth() error {
 
 // sendAndReceiveV0SASLPlainAuth flows the v0 sasl auth NOT wrapped in the kafka protocol
 func (b *Broker) sendAndReceiveV0SASLPlainAuth() error {
-
-	length := 1 + len(b.conf.Net.SASL.User) + 1 + len(b.conf.Net.SASL.Password)
+	length := len(b.conf.Net.SASL.AuthIdentity) + 1 + len(b.conf.Net.SASL.User) + 1 + len(b.conf.Net.SASL.Password)
 	authBytes := make([]byte, length+4) //4 byte length header + auth data
 	binary.BigEndian.PutUint32(authBytes, uint32(length))
-	copy(authBytes[4:], []byte("\x00"+b.conf.Net.SASL.User+"\x00"+b.conf.Net.SASL.Password))
+	copy(authBytes[4:], []byte(b.conf.Net.SASL.AuthIdentity+"\x00"+b.conf.Net.SASL.User+"\x00"+b.conf.Net.SASL.Password))
 
 	requestTime := time.Now()
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	bytesWritten, err := b.write(authBytes)
 	b.updateOutgoingCommunicationMetrics(bytesWritten)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
 		return err
 	}
@@ -1011,11 +1025,13 @@ func (b *Broker) sendAndReceiveV1SASLPlainAuth() error {
 
 	requestTime := time.Now()
 
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	bytesWritten, err := b.sendSASLPlainAuthClientResponse(correlationID)
-
 	b.updateOutgoingCommunicationMetrics(bytesWritten)
 
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
 		return err
 	}
@@ -1068,16 +1084,18 @@ func (b *Broker) sendAndReceiveSASLOAuth(provider AccessTokenProvider) error {
 // if the broker responds with a challenge, in which case the token is
 // rejected.
 func (b *Broker) sendClientMessage(message []byte) (bool, error) {
-
 	requestTime := time.Now()
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	correlationID := b.correlationID
 
 	bytesWritten, err := b.sendSASLOAuthBearerClientMessage(message, correlationID)
+	b.updateOutgoingCommunicationMetrics(bytesWritten)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		return false, err
 	}
 
-	b.updateOutgoingCommunicationMetrics(bytesWritten)
 	b.correlationID++
 
 	res := &SaslAuthenticateResponse{}
@@ -1108,22 +1126,25 @@ func (b *Broker) sendAndReceiveSASLSCRAMv1() error {
 	msg, err := scramClient.Step("")
 	if err != nil {
 		return fmt.Errorf("failed to advance the SCRAM exchange: %s", err.Error())
-
 	}
 
 	for !scramClient.Done() {
 		requestTime := time.Now()
+		// Will be decremented in updateIncomingCommunicationMetrics (except error)
+		b.addRequestInFlightMetrics(1)
 		correlationID := b.correlationID
 		bytesWritten, err := b.sendSaslAuthenticateRequest(correlationID, []byte(msg))
+		b.updateOutgoingCommunicationMetrics(bytesWritten)
 		if err != nil {
+			b.addRequestInFlightMetrics(-1)
 			Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
 			return err
 		}
 
-		b.updateOutgoingCommunicationMetrics(bytesWritten)
 		b.correlationID++
 		challenge, err := b.receiveSaslAuthenticateResponse(correlationID)
 		if err != nil {
+			b.addRequestInFlightMetrics(-1)
 			Logger.Printf("Failed to read response while authenticating with SASL to broker %s: %s\n", b.addr, err.Error())
 			return err
 		}
@@ -1216,7 +1237,7 @@ func mapToString(extensions map[string]string, keyValSep string, elemSep string)
 }
 
 func (b *Broker) sendSASLPlainAuthClientResponse(correlationID int32) (int, error) {
-	authBytes := []byte("\x00" + b.conf.Net.SASL.User + "\x00" + b.conf.Net.SASL.Password)
+	authBytes := []byte(b.conf.Net.SASL.AuthIdentity + "\x00" + b.conf.Net.SASL.User + "\x00" + b.conf.Net.SASL.Password)
 	rb := &SaslAuthenticateRequest{authBytes}
 	req := &request{correlationID: correlationID, clientID: b.conf.ClientID, body: rb}
 	buf, err := encode(req, b.conf.MetricRegistry)
@@ -1228,7 +1249,6 @@ func (b *Broker) sendSASLPlainAuthClientResponse(correlationID int32) (int, erro
 }
 
 func (b *Broker) sendSASLOAuthBearerClientMessage(initialResp []byte, correlationID int32) (int, error) {
-
 	rb := &SaslAuthenticateRequest{initialResp}
 
 	req := &request{correlationID: correlationID, clientID: b.conf.ClientID, body: rb}
@@ -1277,7 +1297,7 @@ func (b *Broker) receiveSASLServerResponse(res *SaslAuthenticateResponse, correl
 }
 
 func (b *Broker) updateIncomingCommunicationMetrics(bytes int, requestLatency time.Duration) {
-	b.updateRequestLatencyMetrics(requestLatency)
+	b.updateRequestLatencyAndInFlightMetrics(requestLatency)
 	b.responseRate.Mark(1)
 
 	if b.brokerResponseRate != nil {
@@ -1296,7 +1316,7 @@ func (b *Broker) updateIncomingCommunicationMetrics(bytes int, requestLatency ti
 	}
 }
 
-func (b *Broker) updateRequestLatencyMetrics(requestLatency time.Duration) {
+func (b *Broker) updateRequestLatencyAndInFlightMetrics(requestLatency time.Duration) {
 	requestLatencyInMs := int64(requestLatency / time.Millisecond)
 	b.requestLatency.Update(requestLatencyInMs)
 
@@ -1304,6 +1324,14 @@ func (b *Broker) updateRequestLatencyMetrics(requestLatency time.Duration) {
 		b.brokerRequestLatency.Update(requestLatencyInMs)
 	}
 
+	b.addRequestInFlightMetrics(-1)
+}
+
+func (b *Broker) addRequestInFlightMetrics(i int64) {
+	b.requestsInFlight.Inc(i)
+	if b.brokerRequestsInFlight != nil {
+		b.brokerRequestsInFlight.Inc(i)
+	}
 }
 
 func (b *Broker) updateOutgoingCommunicationMetrics(bytes int) {
@@ -1322,7 +1350,6 @@ func (b *Broker) updateOutgoingCommunicationMetrics(bytes int) {
 	if b.brokerRequestSize != nil {
 		b.brokerRequestSize.Update(requestSize)
 	}
-
 }
 
 func (b *Broker) registerMetrics() {
@@ -1333,12 +1360,14 @@ func (b *Broker) registerMetrics() {
 	b.brokerOutgoingByteRate = b.registerMeter("outgoing-byte-rate")
 	b.brokerResponseRate = b.registerMeter("response-rate")
 	b.brokerResponseSize = b.registerHistogram("response-size")
+	b.brokerRequestsInFlight = b.registerCounter("requests-in-flight")
 }
 
 func (b *Broker) unregisterMetrics() {
 	for _, name := range b.registeredMetrics {
 		b.conf.MetricRegistry.Unregister(name)
 	}
+	b.registeredMetrics = nil
 }
 
 func (b *Broker) registerMeter(name string) metrics.Meter {
@@ -1352,3 +1381,9 @@ func (b *Broker) registerHistogram(name string) metrics.Histogram {
 	b.registeredMetrics = append(b.registeredMetrics, nameForBroker)
 	return getOrRegisterHistogram(nameForBroker, b.conf.MetricRegistry)
 }
+
+func (b *Broker) registerCounter(name string) metrics.Counter {
+	nameForBroker := getMetricNameForBroker(name, b)
+	b.registeredMetrics = append(b.registeredMetrics, nameForBroker)
+	return metrics.GetOrRegisterCounter(nameForBroker, b.conf.MetricRegistry)
+}
diff --git a/client.go b/client.go
index e5b3557..81a3ca5 100644
--- a/client.go
+++ b/client.go
@@ -17,9 +17,15 @@ type Client interface {
 	// altered after it has been created.
 	Config() *Config
 
-	// Controller returns the cluster controller broker. Requires Kafka 0.10 or higher.
+	// Controller returns the cluster controller broker. It will return a
+	// locally cached value if it's available. You can call RefreshController
+	// to update the cached value. Requires Kafka 0.10 or higher.
 	Controller() (*Broker, error)
 
+	// RefreshController retrieves the cluster controller from fresh metadata
+	// and stores it in the local cache. Requires Kafka 0.10 or higher.
+	RefreshController() (*Broker, error)
+
 	// Brokers returns the current set of active brokers as retrieved from cluster metadata.
 	Brokers() []*Broker
 
@@ -193,7 +199,6 @@ func (client *client) Brokers() []*Broker {
 func (client *client) InitProducerID() (*InitProducerIDResponse, error) {
 	var err error
 	for broker := client.any(); broker != nil; broker = client.any() {
-
 		req := &InitProducerIDRequest{}
 
 		response, err := broker.InitProducerID(req)
@@ -487,6 +492,35 @@ func (client *client) Controller() (*Broker, error) {
 	return controller, nil
 }
 
+// deregisterController removes the cached controllerID
+func (client *client) deregisterController() {
+	client.lock.Lock()
+	defer client.lock.Unlock()
+	delete(client.brokers, client.controllerID)
+}
+
+// RefreshController retrieves the cluster controller from fresh metadata
+// and stores it in the local cache. Requires Kafka 0.10 or higher.
+func (client *client) RefreshController() (*Broker, error) {
+	if client.Closed() {
+		return nil, ErrClosedClient
+	}
+
+	client.deregisterController()
+
+	if err := client.refreshMetadata(); err != nil {
+		return nil, err
+	}
+
+	controller := client.cachedController()
+	if controller == nil {
+		return nil, ErrControllerNotAvailable
+	}
+
+	_ = controller.Open(client.conf)
+	return controller, nil
+}
+
 func (client *client) Coordinator(consumerGroup string) (*Broker, error) {
 	if client.Closed() {
 		return nil, ErrClosedClient
@@ -764,7 +798,7 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 				Logger.Println("client/metadata skipping last retries as we would go past the metadata timeout")
 				return err
 			}
-			Logger.Printf("client/metadata retrying after %dms... (%d attempts remaining)\n", client.conf.Metadata.Retry.Backoff/time.Millisecond, attemptsRemaining)
+			Logger.Printf("client/metadata retrying after %dms... (%d attempts remaining)\n", backoff/time.Millisecond, attemptsRemaining)
 			if backoff > 0 {
 				time.Sleep(backoff)
 			}
diff --git a/compress.go b/compress.go
index 9247c35..12cd7c3 100644
--- a/compress.go
+++ b/compress.go
@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/eapache/go-xerial-snappy"
+	snappy "github.com/eapache/go-xerial-snappy"
 	"github.com/pierrec/lz4"
 )
 
@@ -22,6 +22,87 @@ var (
 			return gzip.NewWriter(nil)
 		},
 	}
+	gzipWriterPoolForCompressionLevel1 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 1)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel2 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 2)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel3 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 3)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel4 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 4)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel5 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 5)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel6 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 6)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel7 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 7)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel8 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 8)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
+	gzipWriterPoolForCompressionLevel9 = sync.Pool{
+		New: func() interface{} {
+			gz, err := gzip.NewWriterLevel(nil, 9)
+			if err != nil {
+				panic(err)
+			}
+			return gz
+		},
+	}
 )
 
 func compress(cc CompressionCodec, level int, data []byte) ([]byte, error) {
@@ -34,15 +115,53 @@ func compress(cc CompressionCodec, level int, data []byte) ([]byte, error) {
 			buf    bytes.Buffer
 			writer *gzip.Writer
 		)
-		if level != CompressionLevelDefault {
+
+		switch level {
+		case CompressionLevelDefault:
+			writer = gzipWriterPool.Get().(*gzip.Writer)
+			defer gzipWriterPool.Put(writer)
+			writer.Reset(&buf)
+		case 1:
+			writer = gzipWriterPoolForCompressionLevel1.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel1.Put(writer)
+			writer.Reset(&buf)
+		case 2:
+			writer = gzipWriterPoolForCompressionLevel2.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel2.Put(writer)
+			writer.Reset(&buf)
+		case 3:
+			writer = gzipWriterPoolForCompressionLevel3.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel3.Put(writer)
+			writer.Reset(&buf)
+		case 4:
+			writer = gzipWriterPoolForCompressionLevel4.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel4.Put(writer)
+			writer.Reset(&buf)
+		case 5:
+			writer = gzipWriterPoolForCompressionLevel5.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel5.Put(writer)
+			writer.Reset(&buf)
+		case 6:
+			writer = gzipWriterPoolForCompressionLevel6.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel6.Put(writer)
+			writer.Reset(&buf)
+		case 7:
+			writer = gzipWriterPoolForCompressionLevel7.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel7.Put(writer)
+			writer.Reset(&buf)
+		case 8:
+			writer = gzipWriterPoolForCompressionLevel8.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel8.Put(writer)
+			writer.Reset(&buf)
+		case 9:
+			writer = gzipWriterPoolForCompressionLevel9.Get().(*gzip.Writer)
+			defer gzipWriterPoolForCompressionLevel9.Put(writer)
+			writer.Reset(&buf)
+		default:
 			writer, err = gzip.NewWriterLevel(&buf, level)
 			if err != nil {
 				return nil, err
 			}
-		} else {
-			writer = gzipWriterPool.Get().(*gzip.Writer)
-			defer gzipWriterPool.Put(writer)
-			writer.Reset(&buf)
 		}
 		if _, err := writer.Write(data); err != nil {
 			return nil, err
diff --git a/config.go b/config.go
index 69c7161..08d50ce 100644
--- a/config.go
+++ b/config.go
@@ -21,6 +21,13 @@ var validID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`)
 type Config struct {
 	// Admin is the namespace for ClusterAdmin properties used by the administrative Kafka client.
 	Admin struct {
+		Retry struct {
+			// The total number of times to retry sending (retriable) admin requests (default 5).
+			// Similar to the `retries` setting of the JVM AdminClientConfig.
+			Max int
+			// Backoff time between retries of a failed request (default 100ms)
+			Backoff time.Duration
+		}
 		// The maximum duration the administrative Kafka client will wait for ClusterAdmin operations,
 		// including topics, brokers, configurations and ACLs (defaults to 3 seconds).
 		Timeout time.Duration
@@ -65,8 +72,15 @@ type Config struct {
 			// (defaults to true). You should only set this to false if you're using
 			// a non-Kafka SASL proxy.
 			Handshake bool
-			//username and password for SASL/PLAIN  or SASL/SCRAM authentication
-			User     string
+			// AuthIdentity is an (optional) authorization identity (authzid) to
+			// use for SASL/PLAIN authentication (if different from User) when
+			// an authenticated user is permitted to act as the presented
+			// alternative user. See RFC4616 for details.
+			AuthIdentity string
+			// User is the authentication identity (authcid) to present for
+			// SASL/PLAIN or SASL/SCRAM authentication
+			User string
+			// Password for SASL/PLAIN authentication
 			Password string
 			// authz id used for SASL/SCRAM authentication
 			SCRAMAuthzID string
@@ -338,6 +352,11 @@ type Config struct {
 		// offsets. This currently requires the manual use of an OffsetManager
 		// but will eventually be automated.
 		Offsets struct {
+			// Deprecated: CommitInterval exists for historical compatibility
+			// and should not be used. Please use Consumer.Offsets.AutoCommit
+			CommitInterval time.Duration
+
+			// AutoCommit specifies configuration for commit messages automatically.
 			AutoCommit struct {
 				// Whether or not to auto-commit updated offsets back to the broker.
 				// (default enabled).
@@ -377,6 +396,10 @@ type Config struct {
 	// debugging, and auditing purposes. Defaults to "sarama", but you should
 	// probably set it to something specific to your application.
 	ClientID string
+	// A rack identifier for this client. This can be any string value which
+	// indicates where this client is physically located.
+	// It corresponds with the broker config 'broker.rack'
+	RackID string
 	// The number of events to buffer in internal and external channels. This
 	// permits the producer and consumer to continue processing some messages
 	// in the background while user code is working, greatly improving throughput.
@@ -401,6 +424,8 @@ type Config struct {
 func NewConfig() *Config {
 	c := &Config{}
 
+	c.Admin.Retry.Max = 5
+	c.Admin.Retry.Backoff = 100 * time.Millisecond
 	c.Admin.Timeout = 3 * time.Second
 
 	c.Net.MaxOpenRequests = 5
@@ -629,6 +654,10 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if c.Producer.Compression == CompressionZSTD && !c.Version.IsAtLeast(V2_1_0_0) {
+		return ConfigurationError("zstd compression requires Version >= V2_1_0_0")
+	}
+
 	if c.Producer.Idempotent {
 		if !c.Version.IsAtLeast(V0_11_0_0) {
 			return ConfigurationError("Idempotent producer requires Version >= V0_11_0_0")
@@ -659,7 +688,7 @@ func (c *Config) Validate() error {
 	case c.Consumer.Retry.Backoff < 0:
 		return ConfigurationError("Consumer.Retry.Backoff must be >= 0")
 	case c.Consumer.Offsets.AutoCommit.Interval <= 0:
-		return ConfigurationError("Consumer.Offsets.CommitInterval must be > 0")
+		return ConfigurationError("Consumer.Offsets.AutoCommit.Interval must be > 0")
 	case c.Consumer.Offsets.Initial != OffsetOldest && c.Consumer.Offsets.Initial != OffsetNewest:
 		return ConfigurationError("Consumer.Offsets.Initial must be OffsetOldest or OffsetNewest")
 	case c.Consumer.Offsets.Retry.Max < 0:
@@ -668,6 +697,11 @@ func (c *Config) Validate() error {
 		return ConfigurationError("Consumer.IsolationLevel must be ReadUncommitted or ReadCommitted")
 	}
 
+	if c.Consumer.Offsets.CommitInterval != 0 {
+		Logger.Println("Deprecation warning: Consumer.Offsets.CommitInterval exists for historical compatibility" +
+			" and should not be used. Please use Consumer.Offsets.AutoCommit, the current value will be ignored")
+	}
+
 	// validate IsolationLevel
 	if c.Consumer.IsolationLevel == ReadCommitted && !c.Version.IsAtLeast(V0_11_0_0) {
 		return ConfigurationError("ReadCommitted requires Version >= V0_11_0_0")
diff --git a/config_resource_type.go b/config_resource_type.go
index 5399d75..bef1053 100644
--- a/config_resource_type.go
+++ b/config_resource_type.go
@@ -1,22 +1,18 @@
 package sarama
 
-//ConfigResourceType is a type for config resource
+// ConfigResourceType is a type for resources that have configs.
 type ConfigResourceType int8
 
-// Taken from :
-// https://cwiki.apache.org/confluence/display/KAFKA/KIP-133%3A+Describe+and+Alter+Configs+Admin+APIs#KIP-133:DescribeandAlterConfigsAdminAPIs-WireFormattypes
+// Taken from:
+// https://github.com/apache/kafka/blob/ed7c071e07f1f90e4c2895582f61ca090ced3c42/clients/src/main/java/org/apache/kafka/common/config/ConfigResource.java#L32-L55
 
 const (
-	//UnknownResource constant type
-	UnknownResource ConfigResourceType = iota
-	//AnyResource constant type
-	AnyResource
-	//TopicResource constant type
-	TopicResource
-	//GroupResource constant type
-	GroupResource
-	//ClusterResource constant type
-	ClusterResource
-	//BrokerResource constant type
-	BrokerResource
+	// UnknownResource constant type
+	UnknownResource ConfigResourceType = 0
+	// TopicResource constant type
+	TopicResource ConfigResourceType = 2
+	// BrokerResource constant type
+	BrokerResource ConfigResourceType = 4
+	// BrokerLoggerResource constant type
+	BrokerLoggerResource ConfigResourceType = 8
 )
diff --git a/consumer.go b/consumer.go
index 72c4d7c..e311c7e 100644
--- a/consumer.go
+++ b/consumer.go
@@ -887,6 +887,13 @@ func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
 		request.Version = 4
 		request.Isolation = bc.consumer.conf.Consumer.IsolationLevel
 	}
+	if bc.consumer.conf.Version.IsAtLeast(V2_1_0_0) {
+		request.Version = 10
+	}
+	if bc.consumer.conf.Version.IsAtLeast(V2_3_0_0) {
+		request.Version = 11
+		request.RackID = bc.consumer.conf.RackID
+	}
 
 	for child := range bc.subscriptions {
 		request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize)
diff --git a/consumer_group.go b/consumer_group.go
index b974dd9..0bc32b9 100644
--- a/consumer_group.go
+++ b/consumer_group.go
@@ -38,6 +38,9 @@ type ConsumerGroup interface {
 	// as quickly as possible to allow time for Cleanup() and the final offset commit. If the timeout
 	// is exceeded, the consumer will be removed from the group by Kafka, which will cause offset
 	// commit failures.
+	// This method should be called inside an infinite loop, when a
+	// server-side rebalance happens, the consumer session will need to be
+	// recreated to get the new claims.
 	Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error
 
 	// Errors returns a read channel of errors that occurred during the consumer life-cycle.
@@ -120,9 +123,6 @@ func (c *consumerGroup) Close() (err error) {
 	c.closeOnce.Do(func() {
 		close(c.closed)
 
-		c.lock.Lock()
-		defer c.lock.Unlock()
-
 		// leave group
 		if e := c.leave(); e != nil {
 			err = e
@@ -175,6 +175,7 @@ func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler Co
 
 	// loop check topic partition numbers changed
 	// will trigger rebalance when any topic partitions number had changed
+	// avoid Consume function called again that will generate more than loopCheckPartitionNumbers coroutine
 	go c.loopCheckPartitionNumbers(topics, sess)
 
 	// Wait for session exit signal
@@ -333,20 +334,14 @@ func (c *consumerGroup) syncGroupRequest(coordinator *Broker, plan BalanceStrate
 		MemberId:     c.memberID,
 		GenerationId: generationID,
 	}
+	strategy := c.config.Consumer.Group.Rebalance.Strategy
 	for memberID, topics := range plan {
 		assignment := &ConsumerGroupMemberAssignment{Topics: topics}
-
-		// Include topic assignments in group-assignment userdata for each consumer-group member
-		if c.config.Consumer.Group.Rebalance.Strategy.Name() == StickyBalanceStrategyName {
-			userDataBytes, err := encode(&StickyAssignorUserDataV1{
-				Topics:     topics,
-				Generation: generationID,
-			}, nil)
-			if err != nil {
-				return nil, err
-			}
-			assignment.UserData = userDataBytes
+		userDataBytes, err := strategy.AssignmentData(memberID, topics, generationID)
+		if err != nil {
+			return nil, err
 		}
+		assignment.UserData = userDataBytes
 		if err := req.AddGroupAssignmentMember(memberID, assignment); err != nil {
 			return nil, err
 		}
@@ -384,8 +379,10 @@ func (c *consumerGroup) balance(members map[string]ConsumerGroupMemberMetadata)
 	return strategy.Plan(members, topics)
 }
 
-// Leaves the cluster, called by Close, protected by lock.
+// Leaves the cluster, called by Close.
 func (c *consumerGroup) leave() error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if c.memberID == "" {
 		return nil
 	}
@@ -430,9 +427,6 @@ func (c *consumerGroup) handleError(err error, topic string, partition int32) {
 		return
 	}
 
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
 	select {
 	case <-c.closed:
 		//consumer is closed
@@ -448,7 +442,7 @@ func (c *consumerGroup) handleError(err error, topic string, partition int32) {
 }
 
 func (c *consumerGroup) loopCheckPartitionNumbers(topics []string, session *consumerGroupSession) {
-	pause := time.NewTicker(c.config.Consumer.Group.Heartbeat.Interval * 2)
+	pause := time.NewTicker(c.config.Metadata.RefreshFrequency)
 	defer session.cancel()
 	defer pause.Stop()
 	var oldTopicToPartitionNum map[string]int
@@ -468,6 +462,10 @@ func (c *consumerGroup) loopCheckPartitionNumbers(topics []string, session *cons
 		}
 		select {
 		case <-pause.C:
+		case <-session.ctx.Done():
+			Logger.Printf("loop check partition number coroutine will exit, topics %s", topics)
+			// if session closed by other, should be exited
+			return
 		case <-c.closed:
 			return
 		}
@@ -475,10 +473,6 @@ func (c *consumerGroup) loopCheckPartitionNumbers(topics []string, session *cons
 }
 
 func (c *consumerGroup) topicToPartitionNumbers(topics []string) (map[string]int, error) {
-	if err := c.client.RefreshMetadata(topics...); err != nil {
-		Logger.Printf("Consumer Group refresh metadata failed %v", err)
-		return nil, err
-	}
 	topicToPartitionNum := make(map[string]int, len(topics))
 	for _, topic := range topics {
 		if partitionNum, err := c.client.Partitions(topic); err != nil {
diff --git a/decompress.go b/decompress.go
index eaccbfc..e4dc3c1 100644
--- a/decompress.go
+++ b/decompress.go
@@ -7,7 +7,7 @@ import (
 	"io/ioutil"
 	"sync"
 
-	"github.com/eapache/go-xerial-snappy"
+	snappy "github.com/eapache/go-xerial-snappy"
 	"github.com/pierrec/lz4"
 )
 
diff --git a/describe_configs_response.go b/describe_configs_response.go
index 5737232..dd919f1 100644
--- a/describe_configs_response.go
+++ b/describe_configs_response.go
@@ -249,12 +249,16 @@ func (r *ConfigEntry) decode(pd packetDecoder, version int16) (err error) {
 			return err
 		}
 		r.Default = defaultB
+		if defaultB {
+			r.Source = SourceDefault
+		}
 	} else {
 		source, err := pd.getInt8()
 		if err != nil {
 			return err
 		}
 		r.Source = ConfigSource(source)
+		r.Default = r.Source == SourceDefault
 	}
 
 	sensitive, err := pd.getBool()
@@ -277,7 +281,6 @@ func (r *ConfigEntry) decode(pd packetDecoder, version int16) (err error) {
 			}
 			r.Synonyms[i] = s
 		}
-
 	}
 	return nil
 }
diff --git a/dev.yml b/dev.yml
index 4c030de..6f6807e 100644
--- a/dev.yml
+++ b/dev.yml
@@ -2,7 +2,7 @@ name: sarama
 
 up:
   - go:
-      version: '1.13.4'
+      version: '1.13.7'
 
 commands:
   test:
diff --git a/examples/consumergroup/main.go b/examples/consumergroup/main.go
index ebbceed..776448f 100644
--- a/examples/consumergroup/main.go
+++ b/examples/consumergroup/main.go
@@ -99,6 +99,9 @@ func main() {
 	go func() {
 		defer wg.Done()
 		for {
+			// `Consume` should be called inside an infinite loop, when a
+			// server-side rebalance happens, the consumer session will need to be
+			// recreated to get the new claims
 			if err := client.Consume(ctx, strings.Split(topics, ","), &consumer); err != nil {
 				log.Panicf("Error from consumer: %v", err)
 			}
diff --git a/fetch_request.go b/fetch_request.go
index 4db9ddd..9a3e8dd 100644
--- a/fetch_request.go
+++ b/fetch_request.go
@@ -1,20 +1,41 @@
 package sarama
 
 type fetchRequestBlock struct {
-	fetchOffset int64
-	maxBytes    int32
+	Version            int16
+	currentLeaderEpoch int32
+	fetchOffset        int64
+	logStartOffset     int64
+	maxBytes           int32
 }
 
-func (b *fetchRequestBlock) encode(pe packetEncoder) error {
+func (b *fetchRequestBlock) encode(pe packetEncoder, version int16) error {
+	b.Version = version
+	if b.Version >= 9 {
+		pe.putInt32(b.currentLeaderEpoch)
+	}
 	pe.putInt64(b.fetchOffset)
+	if b.Version >= 5 {
+		pe.putInt64(b.logStartOffset)
+	}
 	pe.putInt32(b.maxBytes)
 	return nil
 }
 
-func (b *fetchRequestBlock) decode(pd packetDecoder) (err error) {
+func (b *fetchRequestBlock) decode(pd packetDecoder, version int16) (err error) {
+	b.Version = version
+	if b.Version >= 9 {
+		if b.currentLeaderEpoch, err = pd.getInt32(); err != nil {
+			return err
+		}
+	}
 	if b.fetchOffset, err = pd.getInt64(); err != nil {
 		return err
 	}
+	if b.Version >= 5 {
+		if b.logStartOffset, err = pd.getInt64(); err != nil {
+			return err
+		}
+	}
 	if b.maxBytes, err = pd.getInt32(); err != nil {
 		return err
 	}
@@ -25,12 +46,16 @@ func (b *fetchRequestBlock) decode(pd packetDecoder) (err error) {
 // https://issues.apache.org/jira/browse/KAFKA-2063 for a discussion of the issues leading up to that.  The KIP is at
 // https://cwiki.apache.org/confluence/display/KAFKA/KIP-74%3A+Add+Fetch+Response+Size+Limit+in+Bytes
 type FetchRequest struct {
-	MaxWaitTime int32
-	MinBytes    int32
-	MaxBytes    int32
-	Version     int16
-	Isolation   IsolationLevel
-	blocks      map[string]map[int32]*fetchRequestBlock
+	MaxWaitTime  int32
+	MinBytes     int32
+	MaxBytes     int32
+	Version      int16
+	Isolation    IsolationLevel
+	SessionID    int32
+	SessionEpoch int32
+	blocks       map[string]map[int32]*fetchRequestBlock
+	forgotten    map[string][]int32
+	RackID       string
 }
 
 type IsolationLevel int8
@@ -50,6 +75,10 @@ func (r *FetchRequest) encode(pe packetEncoder) (err error) {
 	if r.Version >= 4 {
 		pe.putInt8(int8(r.Isolation))
 	}
+	if r.Version >= 7 {
+		pe.putInt32(r.SessionID)
+		pe.putInt32(r.SessionEpoch)
+	}
 	err = pe.putArrayLength(len(r.blocks))
 	if err != nil {
 		return err
@@ -65,17 +94,44 @@ func (r *FetchRequest) encode(pe packetEncoder) (err error) {
 		}
 		for partition, block := range blocks {
 			pe.putInt32(partition)
-			err = block.encode(pe)
+			err = block.encode(pe, r.Version)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	if r.Version >= 7 {
+		err = pe.putArrayLength(len(r.forgotten))
+		if err != nil {
+			return err
+		}
+		for topic, partitions := range r.forgotten {
+			err = pe.putString(topic)
+			if err != nil {
+				return err
+			}
+			err = pe.putArrayLength(len(partitions))
 			if err != nil {
 				return err
 			}
+			for _, partition := range partitions {
+				pe.putInt32(partition)
+			}
 		}
 	}
+	if r.Version >= 11 {
+		err = pe.putString(r.RackID)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
 func (r *FetchRequest) decode(pd packetDecoder, version int16) (err error) {
 	r.Version = version
+
 	if _, err = pd.getInt32(); err != nil {
 		return err
 	}
@@ -97,6 +153,16 @@ func (r *FetchRequest) decode(pd packetDecoder, version int16) (err error) {
 		}
 		r.Isolation = IsolationLevel(isolation)
 	}
+	if r.Version >= 7 {
+		r.SessionID, err = pd.getInt32()
+		if err != nil {
+			return err
+		}
+		r.SessionEpoch, err = pd.getInt32()
+		if err != nil {
+			return err
+		}
+	}
 	topicCount, err := pd.getArrayLength()
 	if err != nil {
 		return err
@@ -121,12 +187,47 @@ func (r *FetchRequest) decode(pd packetDecoder, version int16) (err error) {
 				return err
 			}
 			fetchBlock := &fetchRequestBlock{}
-			if err = fetchBlock.decode(pd); err != nil {
+			if err = fetchBlock.decode(pd, r.Version); err != nil {
 				return err
 			}
 			r.blocks[topic][partition] = fetchBlock
 		}
 	}
+
+	if r.Version >= 7 {
+		forgottenCount, err := pd.getArrayLength()
+		if err != nil {
+			return err
+		}
+		r.forgotten = make(map[string][]int32)
+		for i := 0; i < forgottenCount; i++ {
+			topic, err := pd.getString()
+			if err != nil {
+				return err
+			}
+			partitionCount, err := pd.getArrayLength()
+			if err != nil {
+				return err
+			}
+			r.forgotten[topic] = make([]int32, partitionCount)
+
+			for j := 0; j < partitionCount; j++ {
+				partition, err := pd.getInt32()
+				if err != nil {
+					return err
+				}
+				r.forgotten[topic][j] = partition
+			}
+		}
+	}
+
+	if r.Version >= 11 {
+		r.RackID, err = pd.getString()
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -140,16 +241,28 @@ func (r *FetchRequest) version() int16 {
 
 func (r *FetchRequest) requiredVersion() KafkaVersion {
 	switch r.Version {
+	case 0:
+		return MinVersion
 	case 1:
 		return V0_9_0_0
 	case 2:
 		return V0_10_0_0
 	case 3:
 		return V0_10_1_0
-	case 4:
+	case 4, 5:
 		return V0_11_0_0
+	case 6:
+		return V1_0_0_0
+	case 7:
+		return V1_1_0_0
+	case 8:
+		return V2_0_0_0
+	case 9, 10:
+		return V2_1_0_0
+	case 11:
+		return V2_3_0_0
 	default:
-		return MinVersion
+		return MaxVersion
 	}
 }
 
@@ -158,13 +271,21 @@ func (r *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int
 		r.blocks = make(map[string]map[int32]*fetchRequestBlock)
 	}
 
+	if r.Version >= 7 && r.forgotten == nil {
+		r.forgotten = make(map[string][]int32)
+	}
+
 	if r.blocks[topic] == nil {
 		r.blocks[topic] = make(map[int32]*fetchRequestBlock)
 	}
 
 	tmp := new(fetchRequestBlock)
+	tmp.Version = r.Version
 	tmp.maxBytes = maxBytes
 	tmp.fetchOffset = fetchOffset
+	if r.Version >= 9 {
+		tmp.currentLeaderEpoch = int32(-1)
+	}
 
 	r.blocks[topic][partitionID] = tmp
 }
diff --git a/fetch_response.go b/fetch_response.go
index 3afc187..dc0aeed 100644
--- a/fetch_response.go
+++ b/fetch_response.go
@@ -30,13 +30,15 @@ func (t *AbortedTransaction) encode(pe packetEncoder) (err error) {
 }
 
 type FetchResponseBlock struct {
-	Err                 KError
-	HighWaterMarkOffset int64
-	LastStableOffset    int64
-	AbortedTransactions []*AbortedTransaction
-	Records             *Records // deprecated: use FetchResponseBlock.RecordsSet
-	RecordsSet          []*Records
-	Partial             bool
+	Err                  KError
+	HighWaterMarkOffset  int64
+	LastStableOffset     int64
+	LogStartOffset       int64
+	AbortedTransactions  []*AbortedTransaction
+	PreferredReadReplica int32
+	Records              *Records // deprecated: use FetchResponseBlock.RecordsSet
+	RecordsSet           []*Records
+	Partial              bool
 }
 
 func (b *FetchResponseBlock) decode(pd packetDecoder, version int16) (err error) {
@@ -57,6 +59,13 @@ func (b *FetchResponseBlock) decode(pd packetDecoder, version int16) (err error)
 			return err
 		}
 
+		if version >= 5 {
+			b.LogStartOffset, err = pd.getInt64()
+			if err != nil {
+				return err
+			}
+		}
+
 		numTransact, err := pd.getArrayLength()
 		if err != nil {
 			return err
@@ -75,6 +84,13 @@ func (b *FetchResponseBlock) decode(pd packetDecoder, version int16) (err error)
 		}
 	}
 
+	if version >= 11 {
+		b.PreferredReadReplica, err = pd.getInt32()
+		if err != nil {
+			return err
+		}
+	}
+
 	recordsSize, err := pd.getInt32()
 	if err != nil {
 		return err
@@ -166,6 +182,10 @@ func (b *FetchResponseBlock) encode(pe packetEncoder, version int16) (err error)
 	if version >= 4 {
 		pe.putInt64(b.LastStableOffset)
 
+		if version >= 5 {
+			pe.putInt64(b.LogStartOffset)
+		}
+
 		if err = pe.putArrayLength(len(b.AbortedTransactions)); err != nil {
 			return err
 		}
@@ -176,6 +196,10 @@ func (b *FetchResponseBlock) encode(pe packetEncoder, version int16) (err error)
 		}
 	}
 
+	if version >= 11 {
+		pe.putInt32(b.PreferredReadReplica)
+	}
+
 	pe.push(&lengthField{})
 	for _, records := range b.RecordsSet {
 		err = records.encode(pe)
@@ -200,7 +224,9 @@ func (b *FetchResponseBlock) getAbortedTransactions() []*AbortedTransaction {
 type FetchResponse struct {
 	Blocks        map[string]map[int32]*FetchResponseBlock
 	ThrottleTime  time.Duration
-	Version       int16 // v1 requires 0.9+, v2 requires 0.10+
+	ErrorCode     int16
+	SessionID     int32
+	Version       int16
 	LogAppendTime bool
 	Timestamp     time.Time
 }
@@ -216,6 +242,17 @@ func (r *FetchResponse) decode(pd packetDecoder, version int16) (err error) {
 		r.ThrottleTime = time.Duration(throttle) * time.Millisecond
 	}
 
+	if r.Version >= 7 {
+		r.ErrorCode, err = pd.getInt16()
+		if err != nil {
+			return err
+		}
+		r.SessionID, err = pd.getInt32()
+		if err != nil {
+			return err
+		}
+	}
+
 	numTopics, err := pd.getArrayLength()
 	if err != nil {
 		return err
@@ -258,6 +295,11 @@ func (r *FetchResponse) encode(pe packetEncoder) (err error) {
 		pe.putInt32(int32(r.ThrottleTime / time.Millisecond))
 	}
 
+	if r.Version >= 7 {
+		pe.putInt16(r.ErrorCode)
+		pe.putInt32(r.SessionID)
+	}
+
 	err = pe.putArrayLength(len(r.Blocks))
 	if err != nil {
 		return err
@@ -281,7 +323,6 @@ func (r *FetchResponse) encode(pe packetEncoder) (err error) {
 				return err
 			}
 		}
-
 	}
 	return nil
 }
@@ -296,16 +337,28 @@ func (r *FetchResponse) version() int16 {
 
 func (r *FetchResponse) requiredVersion() KafkaVersion {
 	switch r.Version {
+	case 0:
+		return MinVersion
 	case 1:
 		return V0_9_0_0
 	case 2:
 		return V0_10_0_0
 	case 3:
 		return V0_10_1_0
-	case 4:
+	case 4, 5:
 		return V0_11_0_0
+	case 6:
+		return V1_0_0_0
+	case 7:
+		return V1_1_0_0
+	case 8:
+		return V2_0_0_0
+	case 9, 10:
+		return V2_1_0_0
+	case 11:
+		return V2_3_0_0
 	default:
-		return MinVersion
+		return MaxVersion
 	}
 }
 
diff --git a/go.mod b/go.mod
index 4337c00..1dca1cc 100644
--- a/go.mod
+++ b/go.mod
@@ -5,25 +5,30 @@ go 1.13
 require (
 	github.com/Shopify/toxiproxy v2.1.4+incompatible
 	github.com/davecgh/go-spew v1.1.1
-	github.com/eapache/go-resiliency v1.1.0
+	github.com/eapache/go-resiliency v1.2.0
 	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21
 	github.com/eapache/queue v1.1.0
 	github.com/fortytw2/leaktest v1.3.0
-	github.com/frankban/quicktest v1.4.1 // indirect
+	github.com/frankban/quicktest v1.7.2 // indirect
 	github.com/golang/snappy v0.0.1 // indirect
-	github.com/hashicorp/go-uuid v1.0.1 // indirect
-	github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 // indirect
-	github.com/klauspost/compress v1.8.2
-	github.com/pierrec/lz4 v2.2.6+incompatible
-	github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a
-	github.com/stretchr/testify v1.3.0
+	github.com/google/go-cmp v0.4.0 // indirect
+	github.com/hashicorp/go-uuid v1.0.2 // indirect
+	github.com/jcmturner/gofork v1.0.0 // indirect
+	github.com/klauspost/compress v1.9.8
+	github.com/kr/pretty v0.2.0 // indirect
+	github.com/pierrec/lz4 v2.4.1+incompatible
+	github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563
+	github.com/stretchr/testify v1.4.0
 	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c
 	github.com/xdg/stringprep v1.0.0 // indirect
-	golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5 // indirect
-	golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3
+	golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72 // indirect
+	golang.org/x/net v0.0.0-20200202094626-16171245cfb2
+	golang.org/x/text v0.3.2 // indirect
+	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/jcmturner/aescts.v1 v1.0.1 // indirect
 	gopkg.in/jcmturner/dnsutils.v1 v1.0.1 // indirect
 	gopkg.in/jcmturner/goidentity.v3 v3.0.0 // indirect
-	gopkg.in/jcmturner/gokrb5.v7 v7.2.3
+	gopkg.in/jcmturner/gokrb5.v7 v7.5.0
 	gopkg.in/jcmturner/rpc.v1 v1.1.0 // indirect
+	gopkg.in/yaml.v2 v2.2.8 // indirect
 )
diff --git a/go.sum b/go.sum
index d2f04ee..06ec328 100644
--- a/go.sum
+++ b/go.sum
@@ -1,67 +1,81 @@
-github.com/DataDog/zstd v1.4.0 h1:vhoV+DUHnRZdKW1i5UMjAk2G4JY8wN4ayRfYDNdEhwo=
-github.com/DataDog/zstd v1.4.0/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
 github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/eapache/go-resiliency v1.1.0 h1:1NtRmCAqadE2FN4ZcN6g90TP3uk8cg9rn9eNK2197aU=
-github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
+github.com/eapache/go-resiliency v1.2.0 h1:v7g92e/KSN71Rq7vSThKaWIq68fL4YHvWyiUKorFR1Q=
+github.com/eapache/go-resiliency v1.2.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
-github.com/frankban/quicktest v1.4.1 h1:Wv2VwvNn73pAdFIVUQRXYDFp31lXKbqblIXo/Q5GPSg=
-github.com/frankban/quicktest v1.4.1/go.mod h1:36zfPVQyHxymz4cH7wlDmVwDrJuljRB60qkgn7rorfQ=
+github.com/frankban/quicktest v1.7.2 h1:2QxQoC1TS09S7fhCPsrvqYdvP1H5M1P1ih5ABm3BTYk=
+github.com/frankban/quicktest v1.7.2/go.mod h1:jaStnuzAqU1AJdCO0l53JDCJrVDKcS03DbaAcR7Ks/o=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
-github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
-github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
-github.com/hashicorp/go-uuid v1.0.1 h1:fv1ep09latC32wFoVwnqcnKJGnMSdBanPczbHAYm1BE=
-github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
-github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjLBdCp5PRlCFijNjvcYANOZXzCfXwCM=
-github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
-github.com/klauspost/compress v1.8.1 h1:oygt2ychZFHOB6M9gUgajzgKrwRgHbGC77NwA4COVgI=
-github.com/klauspost/compress v1.8.1/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
-github.com/klauspost/compress v1.8.2 h1:Bx0qjetmNjdFXASH02NSAREKpiaDwkO1DRZ3dV2KCcs=
-github.com/klauspost/compress v1.8.2/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
+github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
+github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
+github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/jcmturner/gofork v1.0.0 h1:J7uCkflzTEhUZ64xqKnkDxq3kzc96ajM1Gli5ktUem8=
+github.com/jcmturner/gofork v1.0.0/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
+github.com/klauspost/compress v1.9.8 h1:VMAMUUOh+gaxKTMk+zqbjsSjsIcUcL/LF4o63i82QyA=
+github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
+github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/pierrec/lz4 v2.2.6+incompatible h1:6aCX4/YZ9v8q69hTyiR7dNLnTA3fgtKHVVW5BCd5Znw=
-github.com/pierrec/lz4 v2.2.6+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4 v2.4.1+incompatible h1:mFe7ttWaflA46Mhqh+jUfjp2qTbPYxLB2/OyBppH9dg=
+github.com/pierrec/lz4 v2.4.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
-github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
+github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 h1:dY6ETXrvDG7Sa4vE8ZQG4yqWg6UnOcbqTAahkV813vQ=
+github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5 h1:bselrhR0Or1vomJZC8ZIjWtbDmn9OYFLX5Ik9alpJpE=
-golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
+golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72 h1:+ELyKg6m8UBf0nPFSqD0mi7zUfwPyXo23HNjMnXPz7w=
+golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw=
 gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo=
 gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM=
 gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q=
 gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI=
 gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4=
-gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc44010=
-gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM=
+gopkg.in/jcmturner/gokrb5.v7 v7.5.0 h1:a9tsXlIDD9SKxotJMK3niV7rPZAJeX2aD/0yg3qlIrg=
+gopkg.in/jcmturner/gokrb5.v7 v7.5.0/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM=
 gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU=
 gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/gssapi_kerberos.go b/gssapi_kerberos.go
index 57f3ecb..98be0f5 100644
--- a/gssapi_kerberos.go
+++ b/gssapi_kerberos.go
@@ -200,7 +200,6 @@ func (krbAuth *GSSAPIKerberosAuth) initSecContext(bytes []byte, kerberosClient K
 
 /* This does the handshake for authorization */
 func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
-
 	kerberosClient, err := krbAuth.NewKerberosClientFunc(krbAuth.Config)
 	if err != nil {
 		Logger.Printf("Kerberos client error: %s", err)
@@ -243,7 +242,7 @@ func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
 		}
 		broker.updateOutgoingCommunicationMetrics(bytesWritten)
 		if krbAuth.step == GSS_API_VERIFY {
-			var bytesRead = 0
+			bytesRead := 0
 			receivedBytes, bytesRead, err = krbAuth.readPackage(broker)
 			requestLatency := time.Since(requestTime)
 			broker.updateIncomingCommunicationMetrics(bytesRead, requestLatency)
diff --git a/message.go b/message.go
index 7c54748..e48566b 100644
--- a/message.go
+++ b/message.go
@@ -85,7 +85,6 @@ func (m *Message) encode(pe packetEncoder) error {
 		payload = m.compressedCache
 		m.compressedCache = nil
 	} else if m.Value != nil {
-
 		payload, err = compress(m.Codec, m.CompressionLevel, m.Value)
 		if err != nil {
 			return err
diff --git a/metadata_response.go b/metadata_response.go
index b2d532e..916992d 100644
--- a/metadata_response.go
+++ b/metadata_response.go
@@ -318,5 +318,4 @@ foundPartition:
 	pmatch.Isr = isr
 	pmatch.OfflineReplicas = offline
 	pmatch.Err = err
-
 }
diff --git a/mockbroker.go b/mockbroker.go
index 4ed46a6..56e3436 100644
--- a/mockbroker.go
+++ b/mockbroker.go
@@ -213,7 +213,7 @@ func (b *MockBroker) isGSSAPI(buffer []byte) bool {
 	return buffer[4] == 0x60 || bytes.Equal(buffer[4:6], []byte{0x05, 0x04})
 }
 
-func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup) {
+func (b *MockBroker) handleRequests(conn io.ReadWriteCloser, idx int, wg *sync.WaitGroup) {
 	defer wg.Done()
 	defer func() {
 		_ = conn.Close()
@@ -235,7 +235,6 @@ func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup)
 	var bytesWritten int
 	var bytesRead int
 	for {
-
 		buffer, err := b.readToBytes(conn)
 		if err != nil {
 			Logger.Printf("*** mockbroker/%d/%d: invalid request: err=%+v, %+v", b.brokerID, idx, err, spew.Sdump(buffer))
@@ -245,7 +244,6 @@ func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup)
 
 		bytesWritten = 0
 		if !b.isGSSAPI(buffer) {
-
 			req, br, err := decodeRequest(bytes.NewReader(buffer))
 			bytesRead = br
 			if err != nil {
@@ -294,7 +292,6 @@ func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup)
 				break
 			}
 			bytesWritten = len(resHeader) + len(encodedRes)
-
 		} else {
 			// GSSAPI is not part of kafka protocol, but is supported for authentication proposes.
 			// Don't support history for this kind of request as is only used for test GSSAPI authentication mechanism
@@ -317,7 +314,6 @@ func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup)
 			b.notifier(bytesRead, bytesWritten)
 		}
 		b.lock.Unlock()
-
 	}
 	Logger.Printf("*** mockbroker/%d/%d: connection closed, err=%v", b.BrokerID(), idx, err)
 }
diff --git a/mockkerberos.go b/mockkerberos.go
index affeb2d..d36649d 100644
--- a/mockkerberos.go
+++ b/mockkerberos.go
@@ -3,6 +3,7 @@ package sarama
 import (
 	"encoding/binary"
 	"encoding/hex"
+
 	"gopkg.in/jcmturner/gokrb5.v7/credentials"
 	"gopkg.in/jcmturner/gokrb5.v7/gssapi"
 	"gopkg.in/jcmturner/gokrb5.v7/iana/keyusage"
@@ -55,7 +56,6 @@ func (h *KafkaGSSAPIHandler) MockKafkaGSSAPI(buffer []byte) []byte {
 }
 
 type MockKerberosClient struct {
-	asReqBytes  string
 	asRepBytes  string
 	ASRep       messages.ASRep
 	credentials *credentials.Credentials
diff --git a/mockresponses.go b/mockresponses.go
index 7dcc93e..6fa72eb 100644
--- a/mockresponses.go
+++ b/mockresponses.go
@@ -667,7 +667,7 @@ func (mr *MockDeleteTopicsResponse) For(reqBody versionedDecoder) encoder {
 	for _, topic := range req.Topics {
 		res.TopicErrorCodes[topic] = ErrNoError
 	}
-	res.Version = int16(req.Version)
+	res.Version = req.Version
 	return res
 }
 
@@ -731,29 +731,82 @@ func NewMockDescribeConfigsResponse(t TestReporter) *MockDescribeConfigsResponse
 
 func (mr *MockDescribeConfigsResponse) For(reqBody versionedDecoder) encoder {
 	req := reqBody.(*DescribeConfigsRequest)
-	res := &DescribeConfigsResponse{}
+	res := &DescribeConfigsResponse{
+		Version: req.Version,
+	}
+
+	includeSynonyms := (req.Version > 0)
+	includeSource := (req.Version > 0)
 
 	for _, r := range req.Resources {
 		var configEntries []*ConfigEntry
 		switch r.Type {
-		case TopicResource:
+		case BrokerResource:
+			configEntries = append(configEntries,
+				&ConfigEntry{
+					Name:     "min.insync.replicas",
+					Value:    "2",
+					ReadOnly: false,
+					Default:  false,
+				},
+			)
+			res.Resources = append(res.Resources, &ResourceResponse{
+				Name:    r.Name,
+				Configs: configEntries,
+			})
+		case BrokerLoggerResource:
 			configEntries = append(configEntries,
-				&ConfigEntry{Name: "max.message.bytes",
-					Value:     "1000000",
-					ReadOnly:  false,
-					Default:   true,
-					Sensitive: false,
-				}, &ConfigEntry{Name: "retention.ms",
-					Value:     "5000",
-					ReadOnly:  false,
-					Default:   false,
-					Sensitive: false,
-				}, &ConfigEntry{Name: "password",
-					Value:     "12345",
-					ReadOnly:  false,
-					Default:   false,
-					Sensitive: true,
-				})
+				&ConfigEntry{
+					Name:     "kafka.controller.KafkaController",
+					Value:    "DEBUG",
+					ReadOnly: false,
+					Default:  false,
+				},
+			)
+			res.Resources = append(res.Resources, &ResourceResponse{
+				Name:    r.Name,
+				Configs: configEntries,
+			})
+		case TopicResource:
+			maxMessageBytes := &ConfigEntry{Name: "max.message.bytes",
+				Value:     "1000000",
+				ReadOnly:  false,
+				Default:   !includeSource,
+				Sensitive: false,
+			}
+			if includeSource {
+				maxMessageBytes.Source = SourceDefault
+			}
+			if includeSynonyms {
+				maxMessageBytes.Synonyms = []*ConfigSynonym{
+					{
+						ConfigName:  "max.message.bytes",
+						ConfigValue: "500000",
+					},
+				}
+			}
+			retentionMs := &ConfigEntry{Name: "retention.ms",
+				Value:     "5000",
+				ReadOnly:  false,
+				Default:   false,
+				Sensitive: false,
+			}
+			if includeSynonyms {
+				retentionMs.Synonyms = []*ConfigSynonym{
+					{
+						ConfigName:  "log.retention.ms",
+						ConfigValue: "2500",
+					},
+				}
+			}
+			password := &ConfigEntry{Name: "password",
+				Value:     "12345",
+				ReadOnly:  false,
+				Default:   false,
+				Sensitive: true,
+			}
+			configEntries = append(
+				configEntries, maxMessageBytes, retentionMs, password)
 			res.Resources = append(res.Resources, &ResourceResponse{
 				Name:    r.Name,
 				Configs: configEntries,
@@ -777,7 +830,7 @@ func (mr *MockAlterConfigsResponse) For(reqBody versionedDecoder) encoder {
 
 	for _, r := range req.Resources {
 		res.Resources = append(res.Resources, &AlterConfigsResourceResponse{Name: r.Name,
-			Type:     TopicResource,
+			Type:     r.Type,
 			ErrorMsg: "",
 		})
 	}
diff --git a/mocks/async_producer.go b/mocks/async_producer.go
index 488ef07..cbd911c 100644
--- a/mocks/async_producer.go
+++ b/mocks/async_producer.go
@@ -33,7 +33,7 @@ func NewAsyncProducer(t ErrorReporter, config *sarama.Config) *AsyncProducer {
 	}
 	mp := &AsyncProducer{
 		t:            t,
-		closed:       make(chan struct{}, 0),
+		closed:       make(chan struct{}),
 		expectations: make([]*producerExpectation, 0),
 		input:        make(chan *sarama.ProducerMessage, config.ChannelBufferSize),
 		successes:    make(chan *sarama.ProducerMessage, config.ChannelBufferSize),
diff --git a/mocks/mocks.go b/mocks/mocks.go
index 4adb838..1e1d9f1 100644
--- a/mocks/mocks.go
+++ b/mocks/mocks.go
@@ -15,8 +15,6 @@ package mocks
 
 import (
 	"errors"
-
-	"github.com/Shopify/sarama"
 )
 
 // ErrorReporter is a simple interface that includes the testing.T methods we use to report
@@ -41,8 +39,3 @@ type producerExpectation struct {
 	Result        error
 	CheckFunction ValueChecker
 }
-
-type consumerExpectation struct {
-	Err error
-	Msg *sarama.ConsumerMessage
-}
diff --git a/offset_manager.go b/offset_manager.go
index e40f429..1940872 100644
--- a/offset_manager.go
+++ b/offset_manager.go
@@ -280,7 +280,6 @@ func (om *offsetManager) constructRequest() *OffsetCommitRequest {
 			ConsumerID:              om.memberID,
 			ConsumerGroupGeneration: om.generation,
 		}
-
 	}
 
 	om.pomsLock.RLock()
diff --git a/offset_request.go b/offset_request.go
index 326c372..58e2237 100644
--- a/offset_request.go
+++ b/offset_request.go
@@ -6,7 +6,7 @@ type offsetRequestBlock struct {
 }
 
 func (b *offsetRequestBlock) encode(pe packetEncoder, version int16) error {
-	pe.putInt64(int64(b.time))
+	pe.putInt64(b.time)
 	if version == 0 {
 		pe.putInt32(b.maxOffsets)
 	}
diff --git a/produce_request.go b/produce_request.go
index 0c755d0..178972a 100644
--- a/produce_request.go
+++ b/produce_request.go
@@ -214,6 +214,8 @@ func (r *ProduceRequest) requiredVersion() KafkaVersion {
 		return V0_10_0_0
 	case 3:
 		return V0_11_0_0
+	case 7:
+		return V2_1_0_0
 	default:
 		return MinVersion
 	}
diff --git a/produce_response.go b/produce_response.go
index 4c5cd35..7fc2fff 100644
--- a/produce_response.go
+++ b/produce_response.go
@@ -5,11 +5,27 @@ import (
 	"time"
 )
 
+// Protocol, http://kafka.apache.org/protocol.html
+// v1
+// v2 = v3 = v4
+// v5 = v6 = v7
+// Produce Response (Version: 7) => [responses] throttle_time_ms
+//   responses => topic [partition_responses]
+//     topic => STRING
+//     partition_responses => partition error_code base_offset log_append_time log_start_offset
+//       partition => INT32
+//       error_code => INT16
+//       base_offset => INT64
+//       log_append_time => INT64
+//       log_start_offset => INT64
+//   throttle_time_ms => INT32
+
+// partition_responses in protocol
 type ProduceResponseBlock struct {
-	Err    KError
-	Offset int64
-	// only provided if Version >= 2 and the broker is configured with `LogAppendTime`
-	Timestamp time.Time
+	Err         KError    // v0, error_code
+	Offset      int64     // v0, base_offset
+	Timestamp   time.Time // v2, log_append_time, and the broker is configured with `LogAppendTime`
+	StartOffset int64     // v5, log_start_offset
 }
 
 func (b *ProduceResponseBlock) decode(pd packetDecoder, version int16) (err error) {
@@ -32,6 +48,13 @@ func (b *ProduceResponseBlock) decode(pd packetDecoder, version int16) (err erro
 		}
 	}
 
+	if version >= 5 {
+		b.StartOffset, err = pd.getInt64()
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -49,13 +72,17 @@ func (b *ProduceResponseBlock) encode(pe packetEncoder, version int16) (err erro
 		pe.putInt64(timestamp)
 	}
 
+	if version >= 5 {
+		pe.putInt64(b.StartOffset)
+	}
+
 	return nil
 }
 
 type ProduceResponse struct {
-	Blocks       map[string]map[int32]*ProduceResponseBlock
+	Blocks       map[string]map[int32]*ProduceResponseBlock // v0, responses
 	Version      int16
-	ThrottleTime time.Duration // only provided if Version >= 1
+	ThrottleTime time.Duration // v1, throttle_time_ms
 }
 
 func (r *ProduceResponse) decode(pd packetDecoder, version int16) (err error) {
@@ -129,33 +156,13 @@ func (r *ProduceResponse) encode(pe packetEncoder) error {
 			}
 		}
 	}
+
 	if r.Version >= 1 {
 		pe.putInt32(int32(r.ThrottleTime / time.Millisecond))
 	}
 	return nil
 }
 
-func (r *ProduceResponse) key() int16 {
-	return 0
-}
-
-func (r *ProduceResponse) version() int16 {
-	return r.Version
-}
-
-func (r *ProduceResponse) requiredVersion() KafkaVersion {
-	switch r.Version {
-	case 1:
-		return V0_9_0_0
-	case 2:
-		return V0_10_0_0
-	case 3:
-		return V0_11_0_0
-	default:
-		return MinVersion
-	}
-}
-
 func (r *ProduceResponse) GetBlock(topic string, partition int32) *ProduceResponseBlock {
 	if r.Blocks == nil {
 		return nil
diff --git a/produce_set.go b/produce_set.go
index b684aa4..36c43c6 100644
--- a/produce_set.go
+++ b/produce_set.go
@@ -129,6 +129,10 @@ func (ps *produceSet) buildRequest() *ProduceRequest {
 		req.Version = 3
 	}
 
+	if ps.parent.conf.Producer.Compression == CompressionZSTD && ps.parent.conf.Version.IsAtLeast(V2_1_0_0) {
+		req.Version = 7
+	}
+
 	for topic, partitionSets := range ps.msgs {
 		for partition, set := range partitionSets {
 			if req.Version >= 3 {
diff --git a/real_decoder.go b/real_decoder.go
index 085cbb3..6c5a1b9 100644
--- a/real_decoder.go
+++ b/real_decoder.go
@@ -7,9 +7,7 @@ import (
 
 var errInvalidArrayLength = PacketDecodingError{"invalid array length"}
 var errInvalidByteSliceLength = PacketDecodingError{"invalid byteslice length"}
-var errInvalidByteSliceLengthType = PacketDecodingError{"invalid byteslice length type"}
 var errInvalidStringLength = PacketDecodingError{"invalid string length"}
-var errInvalidSubsetSize = PacketDecodingError{"invalid subset size"}
 var errVarintOverflow = PacketDecodingError{"varint overflow"}
 var errInvalidBool = PacketDecodingError{"invalid bool"}
 
diff --git a/records.go b/records.go
index 98160c7..f4c5e95 100644
--- a/records.go
+++ b/records.go
@@ -8,7 +8,6 @@ const (
 	defaultRecords
 
 	magicOffset = 16
-	magicLength = 1
 )
 
 // Records implements a union type containing either a RecordBatch or a legacy MessageSet.
diff --git a/request.go b/request.go
index 97437d6..6e4ad87 100644
--- a/request.go
+++ b/request.go
@@ -105,7 +105,7 @@ func allocateBody(key, version int16) protocolBody {
 	case 0:
 		return &ProduceRequest{}
 	case 1:
-		return &FetchRequest{}
+		return &FetchRequest{Version: version}
 	case 2:
 		return &OffsetRequest{Version: version}
 	case 3:
diff --git a/sarama.go b/sarama.go
index 1e0277a..48f362d 100644
--- a/sarama.go
+++ b/sarama.go
@@ -39,6 +39,10 @@ Broker related metrics:
 	| 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                                            |
 	+----------------------------------------------+------------+---------------------------------------------------------------+
 
 Note that we do not gather specific metrics for seed brokers but they are part of the "all brokers" metrics.
diff --git a/tools/kafka-console-consumer/kafka-console-consumer.go b/tools/kafka-console-consumer/kafka-console-consumer.go
index 824801c..8434665 100644
--- a/tools/kafka-console-consumer/kafka-console-consumer.go
+++ b/tools/kafka-console-consumer/kafka-console-consumer.go
@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
@@ -85,7 +86,7 @@ func main() {
 
 	go func() {
 		signals := make(chan os.Signal, 1)
-		signal.Notify(signals, os.Kill, os.Interrupt)
+		signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
 		<-signals
 		logger.Println("Initiating shutdown of consumer...")
 		close(closing)
diff --git a/tools/kafka-console-partitionconsumer/kafka-console-partitionconsumer.go b/tools/kafka-console-partitionconsumer/kafka-console-partitionconsumer.go
index d5e4464..5873ccd 100644
--- a/tools/kafka-console-partitionconsumer/kafka-console-partitionconsumer.go
+++ b/tools/kafka-console-partitionconsumer/kafka-console-partitionconsumer.go
@@ -8,6 +8,7 @@ import (
 	"os/signal"
 	"strconv"
 	"strings"
+	"syscall"
 
 	"github.com/Shopify/sarama"
 )
@@ -70,7 +71,7 @@ func main() {
 
 	go func() {
 		signals := make(chan os.Signal, 1)
-		signal.Notify(signals, os.Kill, os.Interrupt)
+		signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
 		<-signals
 		pc.AsyncClose()
 	}()
diff --git a/tools/kafka-console-producer/kafka-console-producer.go b/tools/kafka-console-producer/kafka-console-producer.go
index f2b672d..b88e1f1 100644
--- a/tools/kafka-console-producer/kafka-console-producer.go
+++ b/tools/kafka-console-producer/kafka-console-producer.go
@@ -8,13 +8,15 @@ import (
 	"os"
 	"strings"
 
+	"github.com/rcrowley/go-metrics"
+
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
-	"github.com/rcrowley/go-metrics"
 )
 
 var (
 	brokerList    = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The comma separated list of brokers in the Kafka cluster. You can also set the KAFKA_PEERS environment variable")
+	headers       = flag.String("headers", "", "The headers of the message to produce. Example: -headers=foo:bar,bar:foo")
 	topic         = flag.String("topic", "", "REQUIRED: the topic to produce to")
 	key           = flag.String("key", "", "The key of the message to produce. Can be empty.")
 	value         = flag.String("value", "", "REQUIRED: the value of the message to produce. You can also provide the value on stdin.")
@@ -99,6 +101,25 @@ func main() {
 		printUsageErrorAndExit("-value is required, or you have to provide the value on stdin")
 	}
 
+	if *headers != "" {
+		hdrs := []sarama.RecordHeader{}
+		arrHdrs := strings.Split(*headers, ",")
+		for _, h := range arrHdrs {
+			if header := strings.Split(h, ":"); len(header) != 2 {
+				printUsageErrorAndExit("-header should be key:value. Example: -headers=foo:bar,bar:foo")
+			} else {
+				hdrs = append(hdrs, sarama.RecordHeader{
+					Key:   []byte(header[0]),
+					Value: []byte(header[1]),
+				})
+			}
+		}
+
+		if len(hdrs) != 0 {
+			message.Headers = hdrs
+		}
+	}
+
 	producer, err := sarama.NewSyncProducer(strings.Split(*brokerList, ","), config)
 	if err != nil {
 		printErrorAndExit(69, "Failed to open Kafka producer: %s", err)
diff --git a/tools/kafka-producer-performance/main.go b/tools/kafka-producer-performance/main.go
index 7bf1531..b155011 100644
--- a/tools/kafka-producer-performance/main.go
+++ b/tools/kafka-producer-performance/main.go
@@ -14,9 +14,10 @@ import (
 	gosync "sync"
 	"time"
 
+	metrics "github.com/rcrowley/go-metrics"
+
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
-	metrics "github.com/rcrowley/go-metrics"
 )
 
 var (
@@ -80,6 +81,11 @@ var (
 		0,
 		"The maximum number of messages to send per second (0 for no limit).",
 	)
+	maxOpenRequests = flag.Int(
+		"max-open-requests",
+		5,
+		"The maximum number of unacknowledged requests the client will send on a single connection before blocking (default: 5).",
+	)
 	maxMessageBytes = flag.Int(
 		"max-message-bytes",
 		1000000,
@@ -125,16 +131,6 @@ var (
 		0,
 		"The maximum number of messages the producer will send in a single request.",
 	)
-	retryMax = flag.Int(
-		"retry-max",
-		3,
-		"The total number of times to retry sending a message.",
-	)
-	retryBackoff = flag.Duration(
-		"retry-backoff",
-		100*time.Millisecond,
-		"The duration the producer will wait for the cluster to settle between retries.",
-	)
 	clientID = flag.String(
 		"client-id",
 		"sarama",
@@ -248,6 +244,7 @@ func main() {
 
 	config := sarama.NewConfig()
 
+	config.Net.MaxOpenRequests = *maxOpenRequests
 	config.Producer.MaxMessageBytes = *maxMessageBytes
 	config.Producer.RequiredAcks = sarama.RequiredAcks(*requiredAcks)
 	config.Producer.Timeout = *timeout
@@ -430,17 +427,20 @@ func printMetrics(w io.Writer, r metrics.Registry) {
 	recordSendRateMetric := r.Get("record-send-rate")
 	requestLatencyMetric := r.Get("request-latency-in-ms")
 	outgoingByteRateMetric := r.Get("outgoing-byte-rate")
+	requestsInFlightMetric := r.Get("requests-in-flight")
 
-	if recordSendRateMetric == nil || requestLatencyMetric == nil || outgoingByteRateMetric == nil {
+	if recordSendRateMetric == nil || requestLatencyMetric == nil || outgoingByteRateMetric == nil ||
+		requestsInFlightMetric == nil {
 		return
 	}
 	recordSendRate := recordSendRateMetric.(metrics.Meter).Snapshot()
 	requestLatency := requestLatencyMetric.(metrics.Histogram).Snapshot()
 	requestLatencyPercentiles := requestLatency.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
 	outgoingByteRate := outgoingByteRateMetric.(metrics.Meter).Snapshot()
+	requestsInFlight := requestsInFlightMetric.(metrics.Counter).Count()
 	fmt.Fprintf(w, "%d records sent, %.1f records/sec (%.2f MiB/sec ingress, %.2f MiB/sec egress), "+
 		"%.1f ms avg latency, %.1f ms stddev, %.1f ms 50th, %.1f ms 75th, "+
-		"%.1f ms 95th, %.1f ms 99th, %.1f ms 99.9th\n",
+		"%.1f ms 95th, %.1f ms 99th, %.1f ms 99.9th, %d total req. in flight\n",
 		recordSendRate.Count(),
 		recordSendRate.RateMean(),
 		recordSendRate.RateMean()*float64(*messageSize)/1024/1024,
@@ -452,6 +452,7 @@ func printMetrics(w io.Writer, r metrics.Registry) {
 		requestLatencyPercentiles[2],
 		requestLatencyPercentiles[3],
 		requestLatencyPercentiles[4],
+		requestsInFlight,
 	)
 }
 
diff --git a/utils.go b/utils.go
index 9392793..75d0f60 100644
--- a/utils.go
+++ b/utils.go
@@ -26,9 +26,7 @@ func (slice int32Slice) Swap(i, j int) {
 
 func dupInt32Slice(input []int32) []int32 {
 	ret := make([]int32, 0, len(input))
-	for _, val := range input {
-		ret = append(ret, val)
-	}
+	ret = append(ret, input...)
 	return ret
 }
 
diff --git a/vagrant/boot_cluster.sh b/vagrant/boot_cluster.sh
index 95e47dd..6ac8bd4 100755
--- a/vagrant/boot_cluster.sh
+++ b/vagrant/boot_cluster.sh
@@ -17,6 +17,25 @@ while ! nc -q 1 localhost 21805 </dev/null; do echo "Waiting"; sleep 1; done
 # Launch and wait for Kafka
 for i in 1 2 3 4 5; do
     KAFKA_PORT=`expr $i + 9090`
-    cd ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT} && bin/kafka-server-start.sh -daemon config/server.properties
+    cd ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT} && bash bin/kafka-server-start.sh -daemon config/server.properties
 done
-while ! nc -q 1 localhost 29095 </dev/null; do echo "Waiting"; sleep 1; done
+ps auxww | grep -i kafka
+
+N=10
+RC=1
+set +x
+printf "Waiting for Kafka to become available."
+for _ in $(seq 1 "$N"); do
+  if nc -z 127.0.0.1 29095 </dev/null; then
+      RC=0
+      break
+  fi
+  printf "."
+  sleep 1
+done
+printf "\n"
+if [ "$RC" -gt 0 ]; then
+  echo 'Error: Kafka failed to startup' >&2
+  find "${KAFKA_INSTALL_ROOT}" -name "server.log" -print0 | xargs tail -256
+  exit ${RC}
+fi
diff --git a/vagrant/install_cluster.sh b/vagrant/install_cluster.sh
index 0a8892f..aa22261 100755
--- a/vagrant/install_cluster.sh
+++ b/vagrant/install_cluster.sh
@@ -16,25 +16,62 @@ rm -f ${KAFKA_INSTALL_ROOT}/toxiproxy
 ln -s ${KAFKA_INSTALL_ROOT}/toxiproxy-${TOXIPROXY_VERSION} ${KAFKA_INSTALL_ROOT}/toxiproxy
 
 for i in 1 2 3 4 5; do
-    ZK_PORT=`expr $i + 2180`
-    ZK_PORT_REAL=`expr $i + 21800`
-    KAFKA_PORT=`expr $i + 9090`
-    KAFKA_PORT_REAL=`expr $i + 29090`
+    ZK_PORT=$((i + 2180))
+    ZK_PORT_REAL=$((i + 21800))
+    KAFKA_PORT=$((i + 9090))
+    KAFKA_PORT_REAL=$((i + 29090))
 
     # unpack kafka
     mkdir -p ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}
     tar xzf ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_VERSION}.tgz -C ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT} --strip-components 1
 
     # broker configuration
-    cp ${REPOSITORY_ROOT}/vagrant/server.properties ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/
-    sed -i s/KAFKAID/${KAFKA_PORT}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-    sed -i s/KAFKAPORT/${KAFKA_PORT_REAL}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-    sed -i s/KAFKA_HOSTNAME/${KAFKA_HOSTNAME}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-    sed -i s/ZK_PORT/${ZK_PORT}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-
-    KAFKA_DATADIR="${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/data"
-    mkdir -p ${KAFKA_DATADIR}
-    sed -i s#KAFKA_DATADIR#${KAFKA_DATADIR}#g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
+    mkdir -p "${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/data"
+
+    # Append to default server.properties with a small number of customisations
+    printf "\n\n" >> "${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties"
+    cat << EOF >> "${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties"
+############################# Sarama Test Cluster #############################
+
+broker.id=${KAFKA_PORT}
+broker.rack=${i}
+
+# Listen on "real" port
+listeners=PLAINTEXT://:${KAFKA_PORT_REAL}
+# Advertise Toxiproxy port
+advertised.listeners=PLAINTEXT://${KAFKA_HOSTNAME}:${KAFKA_PORT}
+
+# Connect to Zookeeper via Toxiproxy port
+zookeeper.connect=127.0.0.1:${ZK_PORT}
+
+# Data directory
+log.dirs="${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/data"
+
+# Create new topics with a replication factor of 2 so failover can be tested
+# more easily.
+default.replication.factor=2
+
+# Turn on log.retention.bytes to avoid filling up the VM's disk
+log.retention.bytes=268435456
+log.segment.bytes=268435456
+
+# Enable topic deletion and disable auto-creation
+delete.topic.enable=true
+auto.create.topics.enable=false
+
+# Lower the zookeeper timeouts so we don't have to wait forever for a node
+# to die when we use toxiproxy to kill its zookeeper connection
+zookeeper.session.timeout.ms=3000
+zookeeper.connection.timeout.ms=3000
+
+# Disable broker ID length constraint
+reserved.broker.max.id=10000
+
+# Permit follower fetching (KIP-392)
+replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector
+
+###############################################################################
+EOF
 
     # zookeeper configuration
     cp ${REPOSITORY_ROOT}/vagrant/zookeeper.properties ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/
diff --git a/vagrant/server.properties b/vagrant/server.properties
deleted file mode 100644
index ca6e604..0000000
--- a/vagrant/server.properties
+++ /dev/null
@@ -1,127 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-# see kafka.server.KafkaConfig for additional details and defaults
-
-############################# Server Basics #############################
-
-# The id of the broker. This must be set to a unique integer for each broker.
-broker.id=KAFKAID
-reserved.broker.max.id=10000
-
-############################# Socket Server Settings #############################
-
-# The port the socket server listens on
-port=KAFKAPORT
-
-# Hostname the broker will bind to. If not set, the server will bind to all interfaces
-host.name=localhost
-
-# Hostname the broker will advertise to producers and consumers. If not set, it uses the
-# value for "host.name" if configured.  Otherwise, it will use the value returned from
-# java.net.InetAddress.getCanonicalHostName().
-advertised.host.name=KAFKA_HOSTNAME
-advertised.port=KAFKAID
-
-# The port to publish to ZooKeeper for clients to use. If this is not set,
-# it will publish the same port that the broker binds to.
-# advertised.port=<port accessible by clients>
-
-# The number of threads handling network requests
-num.network.threads=2
-
-# The number of threads doing disk I/O
-num.io.threads=8
-
-# The send buffer (SO_SNDBUF) used by the socket server
-socket.send.buffer.bytes=1048576
-
-# The receive buffer (SO_RCVBUF) used by the socket server
-socket.receive.buffer.bytes=1048576
-
-# The maximum size of a request that the socket server will accept (protection against OOM)
-socket.request.max.bytes=104857600
-
-
-############################# Log Basics #############################
-
-# A comma seperated list of directories under which to store log files
-log.dirs=KAFKA_DATADIR
-
-# The default number of log partitions per topic. More partitions allow greater
-# parallelism for consumption, but this will also result in more files across
-# the brokers.
-num.partitions=2
-
-# Create new topics with a replication factor of 2 so failover can be tested
-# more easily.
-default.replication.factor=2
-
-auto.create.topics.enable=false
-delete.topic.enable=true
-
-############################# Log Flush Policy #############################
-
-# Messages are immediately written to the filesystem but by default we only fsync() to sync
-# the OS cache lazily. The following configurations control the flush of data to disk.
-# There are a few important trade-offs here:
-#    1. Durability: Unflushed data may be lost if you are not using replication.
-#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
-#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
-# The settings below allow one to configure the flush policy to flush data after a period of time or
-# every N messages (or both). This can be done globally and overridden on a per-topic basis.
-
-# The number of messages to accept before forcing a flush of data to disk
-#log.flush.interval.messages=10000
-
-# The maximum amount of time a message can sit in a log before we force a flush
-#log.flush.interval.ms=1000
-
-############################# Log Retention Policy #############################
-
-# The following configurations control the disposal of log segments. The policy can
-# be set to delete segments after a period of time, or after a given size has accumulated.
-# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
-# from the end of the log.
-
-# The minimum age of a log file to be eligible for deletion
-log.retention.hours=168
-
-# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
-# segments don't drop below log.retention.bytes.
-log.retention.bytes=268435456
-
-# The maximum size of a log segment file. When this size is reached a new log segment will be created.
-log.segment.bytes=268435456
-
-# The interval at which log segments are checked to see if they can be deleted according
-# to the retention policies
-log.retention.check.interval.ms=60000
-
-# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
-# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
-log.cleaner.enable=false
-
-############################# Zookeeper #############################
-
-# Zookeeper connection string (see zookeeper docs for details).
-# This is a comma separated host:port pairs, each corresponding to a zk
-# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
-# You can also append an optional chroot string to the urls to specify the
-# root directory for all kafka znodes.
-zookeeper.connect=localhost:ZK_PORT
-
-# Timeout in ms for connecting to zookeeper
-zookeeper.session.timeout.ms=3000
-zookeeper.connection.timeout.ms=3000
diff --git a/zstd.go b/zstd.go
index 58880e2..7c9951a 100644
--- a/zstd.go
+++ b/zstd.go
@@ -1,8 +1,9 @@
 package sarama
 
 import (
-	"github.com/klauspost/compress/zstd"
 	"sync"
+
+	"github.com/klauspost/compress/zstd"
 )
 
 var (


Test patch:
diff --git a/add_partitions_to_txn_request_test.go b/add_partitions_to_txn_request_test.go
index 76a2eee..f8dfb27 100644
--- a/add_partitions_to_txn_request_test.go
+++ b/add_partitions_to_txn_request_test.go
@@ -19,7 +19,7 @@ func TestAddPartitionsToTxnRequest(t *testing.T) {
 		ProducerID:      8000,
 		ProducerEpoch:   0,
 		TopicPartitions: map[string][]int32{
-			"topic": []int32{1},
+			"topic": {1},
 		},
 	}
 
diff --git a/add_partitions_to_txn_response_test.go b/add_partitions_to_txn_response_test.go
index 7b5b82f..9cf93e8 100644
--- a/add_partitions_to_txn_response_test.go
+++ b/add_partitions_to_txn_response_test.go
@@ -20,7 +20,7 @@ func TestAddPartitionsToTxnResponse(t *testing.T) {
 	resp := &AddPartitionsToTxnResponse{
 		ThrottleTime: 100 * time.Millisecond,
 		Errors: map[string][]*PartitionError{
-			"topic": []*PartitionError{&PartitionError{
+			"topic": {{
 				Err:       ErrInvalidTxnState,
 				Partition: 2,
 			}},
diff --git a/admin_test.go b/admin_test.go
index 419c230..fcbe539 100644
--- a/admin_test.go
+++ b/admin_test.go
@@ -149,7 +149,7 @@ func TestClusterAdminListTopics(t *testing.T) {
 	})
 
 	config := NewConfig()
-	config.Version = V1_0_0_0
+	config.Version = V1_1_0_0
 	admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
 	if err != nil {
 		t.Fatal(err)
@@ -172,7 +172,7 @@ func TestClusterAdminListTopics(t *testing.T) {
 	if found {
 		t.Fatal(errors.New("default topic config entry incorrectly found in response"))
 	}
-	value, _ := topic.ConfigEntries["retention.ms"]
+	value := topic.ConfigEntries["retention.ms"]
 	if value == nil || *value != "5000" {
 		t.Fatal(errors.New("non-default topic config entry not found in response"))
 	}
@@ -488,21 +488,105 @@ func TestClusterAdminDescribeConfig(t *testing.T) {
 		"DescribeConfigsRequest": NewMockDescribeConfigsResponse(t),
 	})
 
+	var tests = []struct {
+		saramaVersion   KafkaVersion
+		requestVersion  int16
+		includeSynonyms bool
+	}{
+		{V1_0_0_0, 0, false},
+		{V1_1_0_0, 1, true},
+		{V1_1_1_0, 1, true},
+		{V2_0_0_0, 2, true},
+	}
+	for _, tt := range tests {
+		config := NewConfig()
+		config.Version = tt.saramaVersion
+		admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			_ = admin.Close()
+		}()
+
+		resource := ConfigResource{
+			Name:        "r1",
+			Type:        TopicResource,
+			ConfigNames: []string{"my_topic"},
+		}
+
+		entries, err := admin.DescribeConfig(resource)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		history := seedBroker.History()
+		describeReq, ok := history[len(history)-1].Request.(*DescribeConfigsRequest)
+		if !ok {
+			t.Fatal("failed to find DescribeConfigsRequest in mockBroker history")
+		}
+
+		if describeReq.Version != tt.requestVersion {
+			t.Fatalf(
+				"requestVersion %v did not match expected %v",
+				describeReq.Version, tt.requestVersion)
+		}
+
+		if len(entries) <= 0 {
+			t.Fatal(errors.New("no resource present"))
+		}
+		if tt.includeSynonyms {
+			if len(entries[0].Synonyms) == 0 {
+				t.Fatal("expected synonyms to have been included")
+			}
+		}
+	}
+}
+
+// TestClusterAdminDescribeBrokerConfig ensures that a describe broker config
+// is sent to the broker in the resource struct, _not_ the controller
+func TestClusterAdminDescribeBrokerConfig(t *testing.T) {
+	controllerBroker := NewMockBroker(t, 1)
+	defer controllerBroker.Close()
+	configBroker := NewMockBroker(t, 2)
+	defer configBroker.Close()
+
+	controllerBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(controllerBroker.BrokerID()).
+			SetBroker(controllerBroker.Addr(), controllerBroker.BrokerID()).
+			SetBroker(configBroker.Addr(), configBroker.BrokerID()),
+	})
+
+	configBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(controllerBroker.BrokerID()).
+			SetBroker(controllerBroker.Addr(), controllerBroker.BrokerID()).
+			SetBroker(configBroker.Addr(), configBroker.BrokerID()),
+		"DescribeConfigsRequest": NewMockDescribeConfigsResponse(t),
+	})
+
 	config := NewConfig()
 	config.Version = V1_0_0_0
-	admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
+	admin, err := NewClusterAdmin(
+		[]string{
+			controllerBroker.Addr(),
+			configBroker.Addr(),
+		}, config)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	resource := ConfigResource{Name: "r1", Type: TopicResource, ConfigNames: []string{"my_topic"}}
-	entries, err := admin.DescribeConfig(resource)
-	if err != nil {
-		t.Fatal(err)
-	}
+	for _, resourceType := range []ConfigResourceType{BrokerResource, BrokerLoggerResource} {
+		resource := ConfigResource{Name: "2", Type: resourceType}
+		entries, err := admin.DescribeConfig(resource)
+		if err != nil {
+			t.Fatal(err)
+		}
 
-	if len(entries) <= 0 {
-		t.Fatal(errors.New("no resource present"))
+		if len(entries) <= 0 {
+			t.Fatal(errors.New("no resource present"))
+		}
 	}
 
 	err = admin.Close()
@@ -531,8 +615,8 @@ func TestClusterAdminAlterConfig(t *testing.T) {
 
 	var value string
 	entries := make(map[string]*string)
-	value = "3"
-	entries["ReplicationFactor"] = &value
+	value = "60000"
+	entries["retention.ms"] = &value
 	err = admin.AlterConfig(TopicResource, "my_topic", entries, false)
 	if err != nil {
 		t.Fatal(err)
@@ -544,6 +628,60 @@ func TestClusterAdminAlterConfig(t *testing.T) {
 	}
 }
 
+func TestClusterAdminAlterBrokerConfig(t *testing.T) {
+	controllerBroker := NewMockBroker(t, 1)
+	defer controllerBroker.Close()
+	configBroker := NewMockBroker(t, 2)
+	defer configBroker.Close()
+
+	controllerBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(controllerBroker.BrokerID()).
+			SetBroker(controllerBroker.Addr(), controllerBroker.BrokerID()).
+			SetBroker(configBroker.Addr(), configBroker.BrokerID()),
+	})
+	configBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(controllerBroker.BrokerID()).
+			SetBroker(controllerBroker.Addr(), controllerBroker.BrokerID()).
+			SetBroker(configBroker.Addr(), configBroker.BrokerID()),
+		"AlterConfigsRequest": NewMockAlterConfigsResponse(t),
+	})
+
+	config := NewConfig()
+	config.Version = V1_0_0_0
+	admin, err := NewClusterAdmin(
+		[]string{
+			controllerBroker.Addr(),
+			configBroker.Addr(),
+		}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var value string
+	entries := make(map[string]*string)
+	value = "3"
+	entries["min.insync.replicas"] = &value
+
+	for _, resourceType := range []ConfigResourceType{BrokerResource, BrokerLoggerResource} {
+		resource := ConfigResource{Name: "2", Type: resourceType}
+		err = admin.AlterConfig(
+			resource.Type,
+			resource.Name,
+			entries,
+			false)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	err = admin.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestClusterAdminCreateAcl(t *testing.T) {
 	seedBroker := NewMockBroker(t, 1)
 	defer seedBroker.Close()
@@ -821,7 +959,6 @@ func TestListConsumerGroups(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-
 }
 
 func TestListConsumerGroupsMultiBroker(t *testing.T) {
@@ -886,7 +1023,6 @@ func TestListConsumerGroupsMultiBroker(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-
 }
 
 func TestListConsumerGroupOffsets(t *testing.T) {
@@ -915,7 +1051,7 @@ func TestListConsumerGroupOffsets(t *testing.T) {
 	}
 
 	response, err := admin.ListConsumerGroupOffsets(group, map[string][]int32{
-		topic: []int32{0},
+		topic: {0},
 	})
 	if err != nil {
 		t.Fatalf("ListConsumerGroupOffsets failed with error %v", err)
@@ -934,7 +1070,6 @@ func TestListConsumerGroupOffsets(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-
 }
 
 func TestDeleteConsumerGroup(t *testing.T) {
@@ -964,5 +1099,52 @@ func TestDeleteConsumerGroup(t *testing.T) {
 	if err != nil {
 		t.Fatalf("DeleteConsumerGroup failed with error %v", err)
 	}
+}
+
+// TestRefreshMetaDataWithDifferentController ensures that the cached
+// controller can be forcibly updated from Metadata by the admin client
+func TestRefreshMetaDataWithDifferentController(t *testing.T) {
+	seedBroker1 := NewMockBroker(t, 1)
+	seedBroker2 := NewMockBroker(t, 2)
+	defer seedBroker1.Close()
+	defer seedBroker2.Close()
+
+	seedBroker1.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(seedBroker1.BrokerID()).
+			SetBroker(seedBroker1.Addr(), seedBroker1.BrokerID()).
+			SetBroker(seedBroker2.Addr(), seedBroker2.BrokerID()),
+	})
 
+	config := NewConfig()
+	config.Version = V1_1_0_0
+
+	client, err := NewClient([]string{seedBroker1.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ca := clusterAdmin{client: client, conf: config}
+
+	if b, _ := ca.Controller(); seedBroker1.BrokerID() != b.ID() {
+		t.Fatalf("expected cached controller to be %d rather than %d",
+			seedBroker1.BrokerID(), b.ID())
+	}
+
+	seedBroker1.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(seedBroker2.BrokerID()).
+			SetBroker(seedBroker1.Addr(), seedBroker1.BrokerID()).
+			SetBroker(seedBroker2.Addr(), seedBroker2.BrokerID()),
+	})
+
+	if b, _ := ca.refreshController(); seedBroker2.BrokerID() != b.ID() {
+		t.Fatalf("expected refreshed controller to be %d rather than %d",
+			seedBroker2.BrokerID(), b.ID())
+	}
+
+	if b, _ := ca.Controller(); seedBroker2.BrokerID() != b.ID() {
+		t.Fatalf("expected cached controller to be %d rather than %d",
+			seedBroker2.BrokerID(), b.ID())
+	}
 }
diff --git a/alter_configs_request_test.go b/alter_configs_request_test.go
index b9407ca..73c8a62 100644
--- a/alter_configs_request_test.go
+++ b/alter_configs_request_test.go
@@ -51,7 +51,7 @@ func TestAlterConfigsRequest(t *testing.T) {
 	configValue := "1000"
 	request = &AlterConfigsRequest{
 		Resources: []*AlterConfigsResource{
-			&AlterConfigsResource{
+			{
 				Type: TopicResource,
 				Name: "foo",
 				ConfigEntries: map[string]*string{
@@ -65,14 +65,14 @@ func TestAlterConfigsRequest(t *testing.T) {
 
 	request = &AlterConfigsRequest{
 		Resources: []*AlterConfigsResource{
-			&AlterConfigsResource{
+			{
 				Type: TopicResource,
 				Name: "foo",
 				ConfigEntries: map[string]*string{
 					"segment.ms": &configValue,
 				},
 			},
-			&AlterConfigsResource{
+			{
 				Type: TopicResource,
 				Name: "bar",
 				ConfigEntries: map[string]*string{
diff --git a/alter_configs_response_test.go b/alter_configs_response_test.go
index 4592028..9f44dd9 100644
--- a/alter_configs_response_test.go
+++ b/alter_configs_response_test.go
@@ -33,7 +33,7 @@ func TestAlterConfigsResponse(t *testing.T) {
 
 	response = &AlterConfigsResponse{
 		Resources: []*AlterConfigsResourceResponse{
-			&AlterConfigsResourceResponse{
+			{
 				ErrorCode: 0,
 				ErrorMsg:  "",
 				Type:      TopicResource,
diff --git a/api_versions_request_test.go b/api_versions_request_test.go
index 5ab4fa7..be86e62 100644
--- a/api_versions_request_test.go
+++ b/api_versions_request_test.go
@@ -7,8 +7,6 @@ var (
 )
 
 func TestApiVersionsRequest(t *testing.T) {
-	var request *ApiVersionsRequest
-
-	request = new(ApiVersionsRequest)
+	request := new(ApiVersionsRequest)
 	testRequest(t, "basic", request, apiVersionRequest)
 }
diff --git a/api_versions_response_test.go b/api_versions_response_test.go
index 675a65a..58b0658 100644
--- a/api_versions_response_test.go
+++ b/api_versions_response_test.go
@@ -13,9 +13,7 @@ var (
 )
 
 func TestApiVersionsResponse(t *testing.T) {
-	var response *ApiVersionsResponse
-
-	response = new(ApiVersionsResponse)
+	response := new(ApiVersionsResponse)
 	testVersionDecodable(t, "no error", response, apiVersionResponse, 0)
 	if response.Err != ErrNoError {
 		t.Error("Decoding error failed: no error expected but found", response.Err)
diff --git a/async_producer_test.go b/async_producer_test.go
index 1cbc071..3de3086 100644
--- a/async_producer_test.go
+++ b/async_producer_test.go
@@ -311,24 +311,7 @@ func TestAsyncProducerFailureRetry(t *testing.T) {
 	closeProducer(t, producer)
 }
 
-type testLogger struct {
-	t *testing.T
-}
-
-func (l *testLogger) Print(v ...interface{}) {
-	l.t.Log(v...)
-}
-
-func (l *testLogger) Printf(format string, v ...interface{}) {
-	l.t.Logf(format, v...)
-}
-
-func (l *testLogger) Println(v ...interface{}) {
-	l.t.Log(v...)
-}
-
 func TestAsyncProducerRecoveryWithRetriesDisabled(t *testing.T) {
-
 	tt := func(t *testing.T, kErr KError) {
 		seedBroker := NewMockBroker(t, 1)
 		leader1 := NewMockBroker(t, 2)
diff --git a/balance_strategy_test.go b/balance_strategy_test.go
index ab48af3..25a51ad 100644
--- a/balance_strategy_test.go
+++ b/balance_strategy_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"bytes"
 	"fmt"
 	"math"
 	"math/rand"
@@ -62,6 +63,26 @@ func TestBalanceStrategyRange(t *testing.T) {
 	}
 }
 
+func TestBalanceStrategyRangeAssignmentData(t *testing.T) {
+	strategy := BalanceStrategyRange
+
+	members := make(map[string]ConsumerGroupMemberMetadata, 2)
+	members["consumer1"] = ConsumerGroupMemberMetadata{
+		Topics: []string{"topic1"},
+	}
+	members["consumer2"] = ConsumerGroupMemberMetadata{
+		Topics: []string{"topic1"},
+	}
+
+	actual, err := strategy.AssignmentData("consumer1", map[string][]int32{"topic1": {0, 1}}, 1)
+	if err != nil {
+		t.Errorf("Error building assignment data: %v", err)
+	}
+	if actual != nil {
+		t.Error("Invalid assignment data returned from AssignmentData")
+	}
+}
+
 func TestBalanceStrategyRoundRobin(t *testing.T) {
 	tests := []struct {
 		members  map[string][]string
@@ -144,7 +165,7 @@ func Test_deserializeTopicPartitionAssignment(t *testing.T) {
 			want: &StickyAssignorUserDataV0{
 				Topics: map[string][]int32{"t03": {5}},
 				topicPartitions: []topicPartitionAssignment{
-					topicPartitionAssignment{
+					{
 						Topic:     "t03",
 						Partition: 5,
 					},
@@ -165,11 +186,11 @@ func Test_deserializeTopicPartitionAssignment(t *testing.T) {
 				Topics:     map[string][]int32{"t06": {0, 4}},
 				Generation: -1,
 				topicPartitions: []topicPartitionAssignment{
-					topicPartitionAssignment{
+					{
 						Topic:     "t06",
 						Partition: 0,
 					},
-					topicPartitionAssignment{
+					{
 						Topic:     "t06",
 						Partition: 4,
 					},
@@ -191,6 +212,26 @@ func Test_deserializeTopicPartitionAssignment(t *testing.T) {
 	}
 }
 
+func TestBalanceStrategyRoundRobinAssignmentData(t *testing.T) {
+	strategy := BalanceStrategyRoundRobin
+
+	members := make(map[string]ConsumerGroupMemberMetadata, 2)
+	members["consumer1"] = ConsumerGroupMemberMetadata{
+		Topics: []string{"topic1"},
+	}
+	members["consumer2"] = ConsumerGroupMemberMetadata{
+		Topics: []string{"topic1"},
+	}
+
+	actual, err := strategy.AssignmentData("consumer1", map[string][]int32{"topic1": {0, 1}}, 1)
+	if err != nil {
+		t.Errorf("Error building assignment data: %v", err)
+	}
+	if actual != nil {
+		t.Error("Invalid assignment data returned from AssignmentData")
+	}
+}
+
 func Test_prepopulateCurrentAssignments(t *testing.T) {
 	type args struct {
 		members map[string]ConsumerGroupMemberMetadata
@@ -211,7 +252,7 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 			name: "Single consumer",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"c01": ConsumerGroupMemberMetadata{
+					"c01": {
 						Version: 2,
 						UserData: []byte{
 							0x00, 0x00, 0x00, 0x01, 0x00, 0x03, 0x74, 0x30,
@@ -223,12 +264,12 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 				},
 			},
 			wantCurrentAssignments: map[string][]topicPartitionAssignment{
-				"c01": []topicPartitionAssignment{
-					topicPartitionAssignment{
+				"c01": {
+					{
 						Topic:     "t06",
 						Partition: 0,
 					},
-					topicPartitionAssignment{
+					{
 						Topic:     "t06",
 						Partition: 4,
 					},
@@ -240,7 +281,7 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 			name: "Duplicate consumer assignments in metadata",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"c01": ConsumerGroupMemberMetadata{
+					"c01": {
 						Version: 2,
 						UserData: []byte{
 							0x00, 0x00, 0x00, 0x01, 0x00, 0x03, 0x74, 0x30,
@@ -249,7 +290,7 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 							0xff,
 						},
 					},
-					"c02": ConsumerGroupMemberMetadata{
+					"c02": {
 						Version: 2,
 						UserData: []byte{
 							0x00, 0x00, 0x00, 0x01, 0x00, 0x03, 0x74, 0x30,
@@ -261,12 +302,12 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 				},
 			},
 			wantCurrentAssignments: map[string][]topicPartitionAssignment{
-				"c01": []topicPartitionAssignment{
-					topicPartitionAssignment{
+				"c01": {
+					{
 						Topic:     "t06",
 						Partition: 0,
 					},
-					topicPartitionAssignment{
+					{
 						Topic:     "t06",
 						Partition: 4,
 					},
@@ -278,7 +319,7 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 			name: "Different generations (5, 6) of consumer assignments in metadata",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"c01": ConsumerGroupMemberMetadata{
+					"c01": {
 						Version: 2,
 						UserData: []byte{
 							0x00, 0x00, 0x00, 0x01, 0x00, 0x03, 0x74, 0x30,
@@ -287,7 +328,7 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 							0x05,
 						},
 					},
-					"c02": ConsumerGroupMemberMetadata{
+					"c02": {
 						Version: 2,
 						UserData: []byte{
 							0x00, 0x00, 0x00, 0x01, 0x00, 0x03, 0x74, 0x30,
@@ -299,29 +340,29 @@ func Test_prepopulateCurrentAssignments(t *testing.T) {
 				},
 			},
 			wantCurrentAssignments: map[string][]topicPartitionAssignment{
-				"c01": []topicPartitionAssignment{
-					topicPartitionAssignment{
+				"c01": {
+					{
 						Topic:     "t06",
 						Partition: 0,
 					},
-					topicPartitionAssignment{
+					{
 						Topic:     "t06",
 						Partition: 4,
 					},
 				},
 			},
 			wantPrevAssignments: map[topicPartitionAssignment]consumerGenerationPair{
-				topicPartitionAssignment{
+				{
 					Topic:     "t06",
 					Partition: 0,
-				}: consumerGenerationPair{
+				}: {
 					Generation: 5,
 					MemberID:   "c01",
 				},
-				topicPartitionAssignment{
+				{
 					Topic:     "t06",
 					Partition: 4,
-				}: consumerGenerationPair{
+				}: {
 					Generation: 5,
 					MemberID:   "c01",
 				},
@@ -365,9 +406,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			name: "Topic partitions with identical consumer entries",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 2}: {"c1", "c2", "c3"},
 				},
 				consumer2AllPotentialPartitions: make(map[string][]topicPartitionAssignment),
 			},
@@ -377,9 +418,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			name: "Topic partitions with mixed up consumer entries",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c3", "c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"c3", "c1", "c2"},
 				},
 				consumer2AllPotentialPartitions: make(map[string][]topicPartitionAssignment),
 			},
@@ -389,9 +430,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			name: "Topic partitions with different consumer entries",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"cX", "c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"cX", "c1", "c2"},
 				},
 				consumer2AllPotentialPartitions: make(map[string][]topicPartitionAssignment),
 			},
@@ -401,9 +442,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			name: "Topic partitions with different number of consumer entries",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"c1", "c2"},
 				},
 				consumer2AllPotentialPartitions: make(map[string][]topicPartitionAssignment),
 			},
@@ -414,9 +455,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			args: args{
 				partition2AllPotentialConsumers: make(map[topicPartitionAssignment][]string),
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c3": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
 				},
 			},
 			want: true,
@@ -426,9 +467,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			args: args{
 				partition2AllPotentialConsumers: make(map[topicPartitionAssignment][]string),
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}, topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 2}, topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}, {Topic: "t1", Partition: 0}},
+					"c3": {{Topic: "t1", Partition: 2}, {Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}},
 				},
 			},
 			want: true,
@@ -438,9 +479,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			args: args{
 				partition2AllPotentialConsumers: make(map[topicPartitionAssignment][]string),
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}, topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "tX", Partition: 2}, topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}, {Topic: "t1", Partition: 0}},
+					"c3": {{Topic: "tX", Partition: 2}, {Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}},
 				},
 			},
 			want: false,
@@ -450,9 +491,9 @@ func Test_areSubscriptionsIdentical(t *testing.T) {
 			args: args{
 				partition2AllPotentialConsumers: make(map[topicPartitionAssignment][]string),
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}, topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}, {Topic: "t1", Partition: 0}},
+					"c3": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}},
 				},
 			},
 			want: false,
@@ -484,10 +525,10 @@ func Test_sortMemberIDsByPartitionAssignments(t *testing.T) {
 			name: "Single assignment",
 			args: args{
 				assignments: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
+						{Topic: "t1", Partition: 2},
 					},
 				},
 			},
@@ -497,17 +538,17 @@ func Test_sortMemberIDsByPartitionAssignments(t *testing.T) {
 			name: "Multiple assignments with different partition counts",
 			args: args{
 				assignments: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
+					"c1": {
+						{Topic: "t1", Partition: 0},
 					},
-					"c2": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c2": {
+						{Topic: "t1", Partition: 1},
+						{Topic: "t1", Partition: 2},
 					},
-					"c3": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 3},
-						topicPartitionAssignment{Topic: "t1", Partition: 4},
-						topicPartitionAssignment{Topic: "t1", Partition: 5},
+					"c3": {
+						{Topic: "t1", Partition: 3},
+						{Topic: "t1", Partition: 4},
+						{Topic: "t1", Partition: 5},
 					},
 				},
 			},
@@ -544,19 +585,19 @@ func Test_sortPartitions(t *testing.T) {
 			name: "Base case",
 			args: args{
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}},
+					"c2": {{Topic: "t1", Partition: 1}},
+					"c3": {{Topic: "t1", Partition: 2}},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c3": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c3", "c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"c3", "c1", "c2"},
 				},
 			},
 		},
@@ -564,20 +605,20 @@ func Test_sortPartitions(t *testing.T) {
 			name: "Partitions assigned to a different consumer last time",
 			args: args{
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
+					"c1": {{Topic: "t1", Partition: 0}},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c3": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c3", "c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"c3", "c1", "c2"},
 				},
 				partitionsWithADifferentPreviousAssignment: map[topicPartitionAssignment]consumerGenerationPair{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: consumerGenerationPair{Generation: 1, MemberID: "c2"},
+					{Topic: "t1", Partition: 0}: {Generation: 1, MemberID: "c2"},
 				},
 			},
 		},
@@ -585,21 +626,21 @@ func Test_sortPartitions(t *testing.T) {
 			name: "Partitions assigned to a different consumer last time",
 			args: args{
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}},
+					"c1": {{Topic: "t1", Partition: 0}},
+					"c2": {{Topic: "t1", Partition: 1}},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c3": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c3", "c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"c3", "c1", "c2"},
 				},
 				partitionsWithADifferentPreviousAssignment: map[topicPartitionAssignment]consumerGenerationPair{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: consumerGenerationPair{Generation: 1, MemberID: "c2"},
+					{Topic: "t1", Partition: 0}: {Generation: 1, MemberID: "c2"},
 				},
 			},
 		},
@@ -609,17 +650,17 @@ func Test_sortPartitions(t *testing.T) {
 				isFreshAssignment: true,
 				currentAssignment: map[string][]topicPartitionAssignment{},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}, topicPartitionAssignment{Topic: "t1", Partition: 1}, topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c2": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
+					"c3": {{Topic: "t1", Partition: 0}, {Topic: "t1", Partition: 1}, {Topic: "t1", Partition: 2}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2", "c3"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2", "c3", "c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c3", "c1", "c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2", "c3"},
+					{Topic: "t1", Partition: 1}: {"c2", "c3", "c1"},
+					{Topic: "t1", Partition: 2}: {"c3", "c1", "c2"},
 				},
 				partitionsWithADifferentPreviousAssignment: map[topicPartitionAssignment]consumerGenerationPair{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: consumerGenerationPair{Generation: 1, MemberID: "c2"},
+					{Topic: "t1", Partition: 0}: {Generation: 1, MemberID: "c2"},
 				},
 			},
 		},
@@ -648,52 +689,52 @@ func Test_filterAssignedPartitions(t *testing.T) {
 			name: "All partitions accounted for",
 			args: args{
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}},
+					"c1": {{Topic: "t1", Partition: 0}},
+					"c2": {{Topic: "t1", Partition: 1}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c2"},
+					{Topic: "t1", Partition: 0}: {"c1"},
+					{Topic: "t1", Partition: 1}: {"c2"},
 				},
 			},
 			want: map[string][]topicPartitionAssignment{
-				"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-				"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}},
+				"c1": {{Topic: "t1", Partition: 0}},
+				"c2": {{Topic: "t1", Partition: 1}},
 			},
 		},
 		{
 			name: "One consumer using an unrecognized partition",
 			args: args{
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}},
+					"c1": {{Topic: "t1", Partition: 0}},
+					"c2": {{Topic: "t1", Partition: 1}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1"},
+					{Topic: "t1", Partition: 0}: {"c1"},
 				},
 			},
 			want: map[string][]topicPartitionAssignment{
-				"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-				"c2": []topicPartitionAssignment{},
+				"c1": {{Topic: "t1", Partition: 0}},
+				"c2": {},
 			},
 		},
 		{
 			name: "Interleaved consumer removal",
 			args: args{
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-					"c2": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 1}},
-					"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 2}},
+					"c1": {{Topic: "t1", Partition: 0}},
+					"c2": {{Topic: "t1", Partition: 1}},
+					"c3": {{Topic: "t1", Partition: 2}},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c3"},
+					{Topic: "t1", Partition: 0}: {"c1"},
+					{Topic: "t1", Partition: 2}: {"c3"},
 				},
 			},
 			want: map[string][]topicPartitionAssignment{
-				"c1": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 0}},
-				"c2": []topicPartitionAssignment{},
-				"c3": []topicPartitionAssignment{topicPartitionAssignment{Topic: "t1", Partition: 2}},
+				"c1": {{Topic: "t1", Partition: 0}},
+				"c2": {},
+				"c3": {{Topic: "t1", Partition: 2}},
 			},
 		},
 	}
@@ -723,28 +764,28 @@ func Test_canConsumerParticipateInReassignment(t *testing.T) {
 			args: args{
 				memberID: "c1",
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
+						{Topic: "t1", Partition: 2},
 					},
-					"c2": []topicPartitionAssignment{},
+					"c2": {},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
 					},
-					"c2": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c2": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
+						{Topic: "t1", Partition: 2},
 					},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1", "c2"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c1", "c2"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c2"},
+					{Topic: "t1", Partition: 0}: {"c1", "c2"},
+					{Topic: "t1", Partition: 1}: {"c1", "c2"},
+					{Topic: "t1", Partition: 2}: {"c2"},
 				},
 			},
 			want: true,
@@ -754,20 +795,20 @@ func Test_canConsumerParticipateInReassignment(t *testing.T) {
 			args: args{
 				memberID: "c1",
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
 					},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
 					},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c1"},
+					{Topic: "t1", Partition: 0}: {"c1"},
+					{Topic: "t1", Partition: 1}: {"c1"},
 				},
 			},
 			want: false,
@@ -777,22 +818,22 @@ func Test_canConsumerParticipateInReassignment(t *testing.T) {
 			args: args{
 				memberID: "c1",
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
 					},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c1": {
+						{Topic: "t1", Partition: 0},
+						{Topic: "t1", Partition: 1},
+						{Topic: "t1", Partition: 2},
 					},
 				},
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: []string{"c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: []string{"c1"},
-					topicPartitionAssignment{Topic: "t1", Partition: 2}: []string{"c1"},
+					{Topic: "t1", Partition: 0}: {"c1"},
+					{Topic: "t1", Partition: 1}: {"c1"},
+					{Topic: "t1", Partition: 2}: {"c1"},
 				},
 			},
 			want: true,
@@ -829,45 +870,45 @@ func Test_removeTopicPartitionFromMemberAssignments(t *testing.T) {
 			name: "Remove first entry",
 			args: args{
 				assignments: []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 0},
-					topicPartitionAssignment{Topic: "t1", Partition: 1},
-					topicPartitionAssignment{Topic: "t1", Partition: 2},
+					{Topic: "t1", Partition: 0},
+					{Topic: "t1", Partition: 1},
+					{Topic: "t1", Partition: 2},
 				},
 				topic: topicPartitionAssignment{Topic: "t1", Partition: 0},
 			},
 			want: []topicPartitionAssignment{
-				topicPartitionAssignment{Topic: "t1", Partition: 1},
-				topicPartitionAssignment{Topic: "t1", Partition: 2},
+				{Topic: "t1", Partition: 1},
+				{Topic: "t1", Partition: 2},
 			},
 		},
 		{
 			name: "Remove middle entry",
 			args: args{
 				assignments: []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 0},
-					topicPartitionAssignment{Topic: "t1", Partition: 1},
-					topicPartitionAssignment{Topic: "t1", Partition: 2},
+					{Topic: "t1", Partition: 0},
+					{Topic: "t1", Partition: 1},
+					{Topic: "t1", Partition: 2},
 				},
 				topic: topicPartitionAssignment{Topic: "t1", Partition: 1},
 			},
 			want: []topicPartitionAssignment{
-				topicPartitionAssignment{Topic: "t1", Partition: 0},
-				topicPartitionAssignment{Topic: "t1", Partition: 2},
+				{Topic: "t1", Partition: 0},
+				{Topic: "t1", Partition: 2},
 			},
 		},
 		{
 			name: "Remove last entry",
 			args: args{
 				assignments: []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 0},
-					topicPartitionAssignment{Topic: "t1", Partition: 1},
-					topicPartitionAssignment{Topic: "t1", Partition: 2},
+					{Topic: "t1", Partition: 0},
+					{Topic: "t1", Partition: 1},
+					{Topic: "t1", Partition: 2},
 				},
 				topic: topicPartitionAssignment{Topic: "t1", Partition: 2},
 			},
 			want: []topicPartitionAssignment{
-				topicPartitionAssignment{Topic: "t1", Partition: 0},
-				topicPartitionAssignment{Topic: "t1", Partition: 1},
+				{Topic: "t1", Partition: 0},
+				{Topic: "t1", Partition: 1},
 			},
 		},
 	}
@@ -901,46 +942,46 @@ func Test_assignPartition(t *testing.T) {
 				partition:                  topicPartitionAssignment{Topic: "t1", Partition: 2},
 				sortedCurrentSubscriptions: []string{"c3", "c1", "c2"},
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
+					"c1": {
+						{Topic: "t1", Partition: 0},
 					},
-					"c2": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c2": {
+						{Topic: "t1", Partition: 1},
 					},
-					"c3": []topicPartitionAssignment{},
+					"c3": {},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
+					"c1": {
+						{Topic: "t1", Partition: 0},
 					},
-					"c2": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c2": {
+						{Topic: "t1", Partition: 1},
 					},
-					"c3": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c3": {
+						{Topic: "t1", Partition: 2},
 					},
 				},
 				currentPartitionConsumer: map[topicPartitionAssignment]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: "c1",
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: "c2",
+					{Topic: "t1", Partition: 0}: "c1",
+					{Topic: "t1", Partition: 1}: "c2",
 				},
 			},
 			want: []string{"c1", "c2", "c3"},
 			wantCurrentAssignment: map[string][]topicPartitionAssignment{
-				"c1": []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 0},
+				"c1": {
+					{Topic: "t1", Partition: 0},
 				},
-				"c2": []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 1},
+				"c2": {
+					{Topic: "t1", Partition: 1},
 				},
-				"c3": []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 2},
+				"c3": {
+					{Topic: "t1", Partition: 2},
 				},
 			},
 			wantCurrentPartitionConsumer: map[topicPartitionAssignment]string{
-				topicPartitionAssignment{Topic: "t1", Partition: 0}: "c1",
-				topicPartitionAssignment{Topic: "t1", Partition: 1}: "c2",
-				topicPartitionAssignment{Topic: "t1", Partition: 2}: "c3",
+				{Topic: "t1", Partition: 0}: "c1",
+				{Topic: "t1", Partition: 1}: "c2",
+				{Topic: "t1", Partition: 2}: "c3",
 			},
 		},
 		{
@@ -949,43 +990,43 @@ func Test_assignPartition(t *testing.T) {
 				partition:                  topicPartitionAssignment{Topic: "t1", Partition: 3},
 				sortedCurrentSubscriptions: []string{"c3", "c1", "c2"},
 				currentAssignment: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
+					"c1": {
+						{Topic: "t1", Partition: 0},
 					},
-					"c2": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c2": {
+						{Topic: "t1", Partition: 1},
 					},
-					"c3": []topicPartitionAssignment{},
+					"c3": {},
 				},
 				consumer2AllPotentialPartitions: map[string][]topicPartitionAssignment{
-					"c1": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 0},
+					"c1": {
+						{Topic: "t1", Partition: 0},
 					},
-					"c2": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 1},
+					"c2": {
+						{Topic: "t1", Partition: 1},
 					},
-					"c3": []topicPartitionAssignment{
-						topicPartitionAssignment{Topic: "t1", Partition: 2},
+					"c3": {
+						{Topic: "t1", Partition: 2},
 					},
 				},
 				currentPartitionConsumer: map[topicPartitionAssignment]string{
-					topicPartitionAssignment{Topic: "t1", Partition: 0}: "c1",
-					topicPartitionAssignment{Topic: "t1", Partition: 1}: "c2",
+					{Topic: "t1", Partition: 0}: "c1",
+					{Topic: "t1", Partition: 1}: "c2",
 				},
 			},
 			want: []string{"c3", "c1", "c2"},
 			wantCurrentAssignment: map[string][]topicPartitionAssignment{
-				"c1": []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 0},
+				"c1": {
+					{Topic: "t1", Partition: 0},
 				},
-				"c2": []topicPartitionAssignment{
-					topicPartitionAssignment{Topic: "t1", Partition: 1},
+				"c2": {
+					{Topic: "t1", Partition: 1},
 				},
-				"c3": []topicPartitionAssignment{},
+				"c3": {},
 			},
 			wantCurrentPartitionConsumer: map[topicPartitionAssignment]string{
-				topicPartitionAssignment{Topic: "t1", Partition: 0}: "c1",
-				topicPartitionAssignment{Topic: "t1", Partition: 1}: "c2",
+				{Topic: "t1", Partition: 0}: "c1",
+				{Topic: "t1", Partition: 1}: "c2",
 			},
 		},
 	}
@@ -1018,7 +1059,7 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "One consumer with no topics",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer": ConsumerGroupMemberMetadata{},
+					"consumer": {},
 				},
 				topics: make(map[string][]int32),
 			},
@@ -1027,7 +1068,7 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "One consumer with non-existent topic",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer": ConsumerGroupMemberMetadata{
+					"consumer": {
 						Topics: []string{"topic"},
 					},
 				},
@@ -1040,12 +1081,12 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "One consumer with one topic",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer": ConsumerGroupMemberMetadata{
+					"consumer": {
 						Topics: []string{"topic"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic": []int32{0, 1, 2},
+					"topic": {0, 1, 2},
 				},
 			},
 		},
@@ -1053,13 +1094,13 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Only assigns partitions from subscribed topics",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer": ConsumerGroupMemberMetadata{
+					"consumer": {
 						Topics: []string{"topic"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic": []int32{0, 1, 2},
-					"other": []int32{0, 1, 2},
+					"topic": {0, 1, 2},
+					"other": {0, 1, 2},
 				},
 			},
 		},
@@ -1067,13 +1108,13 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "One consumer with multiple topics",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer": ConsumerGroupMemberMetadata{
+					"consumer": {
 						Topics: []string{"topic1", "topic2"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0},
-					"topic2": []int32{0, 1},
+					"topic1": {0},
+					"topic2": {0, 1},
 				},
 			},
 		},
@@ -1081,15 +1122,15 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Two consumers with one topic and one partition",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics: []string{"topic"},
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics: []string{"topic"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic": []int32{0},
+					"topic": {0},
 				},
 			},
 		},
@@ -1097,15 +1138,15 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Two consumers with one topic and two partitions",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics: []string{"topic"},
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics: []string{"topic"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic": []int32{0, 1},
+					"topic": {0, 1},
 				},
 			},
 		},
@@ -1113,19 +1154,19 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Multiple consumers with mixed topic subscriptions",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics: []string{"topic1"},
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics: []string{"topic1", "topic2"},
 					},
-					"consumer3": ConsumerGroupMemberMetadata{
+					"consumer3": {
 						Topics: []string{"topic1"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0, 1, 2},
-					"topic2": []int32{0, 1},
+					"topic1": {0, 1, 2},
+					"topic2": {0, 1},
 				},
 			},
 		},
@@ -1133,16 +1174,16 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Two consumers with two topics and six partitions",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics: []string{"topic1", "topic2"},
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics: []string{"topic1", "topic2"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0, 1, 2},
-					"topic2": []int32{0, 1, 2},
+					"topic1": {0, 1, 2},
+					"topic2": {0, 1, 2},
 				},
 			},
 		},
@@ -1150,20 +1191,20 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Three consumers (two old, one new) with one topic and twelve partitions",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics:   []string{"topic1"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{4, 11, 8, 5, 9, 2}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {4, 11, 8, 5, 9, 2}}, 1),
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics:   []string{"topic1"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{1, 3, 0, 7, 10, 6}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {1, 3, 0, 7, 10, 6}}, 1),
 					},
-					"consumer3": ConsumerGroupMemberMetadata{
+					"consumer3": {
 						Topics: []string{"topic1"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
+					"topic1": {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
 				},
 			},
 		},
@@ -1171,20 +1212,20 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Three consumers (two old, one new) with one topic and 13 partitions",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics:   []string{"topic1"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{4, 11, 8, 5, 9, 2, 6}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {4, 11, 8, 5, 9, 2, 6}}, 1),
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics:   []string{"topic1"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{1, 3, 0, 7, 10, 12}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {1, 3, 0, 7, 10, 12}}, 1),
 					},
-					"consumer3": ConsumerGroupMemberMetadata{
+					"consumer3": {
 						Topics: []string{"topic1"},
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12},
+					"topic1": {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12},
 				},
 			},
 		},
@@ -1192,14 +1233,14 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "One consumer that is no longer subscribed to a topic that it had previously been consuming from",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics:   []string{"topic2"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0}}, 1),
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0},
-					"topic2": []int32{0},
+					"topic1": {0},
+					"topic2": {0},
 				},
 			},
 		},
@@ -1207,18 +1248,18 @@ func Test_stickyBalanceStrategy_Plan(t *testing.T) {
 			name: "Two consumers where one is no longer interested in consuming from a topic that it had been consuming from",
 			args: args{
 				members: map[string]ConsumerGroupMemberMetadata{
-					"consumer1": ConsumerGroupMemberMetadata{
+					"consumer1": {
 						Topics:   []string{"topic2"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0}}, 1),
 					},
-					"consumer2": ConsumerGroupMemberMetadata{
+					"consumer2": {
 						Topics:   []string{"topic1", "topic2"},
-						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{1}}, 1),
+						UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {1}}, 1),
 					},
 				},
 				topics: map[string][]int32{
-					"topic1": []int32{0, 1},
-					"topic2": []int32{0, 1},
+					"topic1": {0, 1},
+					"topic2": {0, 1},
 				},
 			},
 		},
@@ -1238,21 +1279,21 @@ func Test_stickyBalanceStrategy_Plan_KIP54_ExampleOne(t *testing.T) {
 
 	// PLAN 1
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{
+		"consumer1": {
 			Topics: []string{"topic1", "topic2", "topic3", "topic4"},
 		},
-		"consumer2": ConsumerGroupMemberMetadata{
+		"consumer2": {
 			Topics: []string{"topic1", "topic2", "topic3", "topic4"},
 		},
-		"consumer3": ConsumerGroupMemberMetadata{
+		"consumer3": {
 			Topics: []string{"topic1", "topic2", "topic3", "topic4"},
 		},
 	}
 	topics := map[string][]int32{
-		"topic1": []int32{0, 1},
-		"topic2": []int32{0, 1},
-		"topic3": []int32{0, 1},
-		"topic4": []int32{0, 1},
+		"topic1": {0, 1},
+		"topic2": {0, 1},
+		"topic3": {0, 1},
+		"topic4": {0, 1},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1278,20 +1319,20 @@ func Test_stickyBalanceStrategy_Plan_KIP54_ExampleTwo(t *testing.T) {
 
 	// PLAN 1
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{
+		"consumer1": {
 			Topics: []string{"topic1"},
 		},
-		"consumer2": ConsumerGroupMemberMetadata{
+		"consumer2": {
 			Topics: []string{"topic1", "topic2"},
 		},
-		"consumer3": ConsumerGroupMemberMetadata{
+		"consumer3": {
 			Topics: []string{"topic1", "topic2", "topic3"},
 		},
 	}
 	topics := map[string][]int32{
-		"topic1": []int32{0},
-		"topic2": []int32{0, 1},
-		"topic3": []int32{0, 1, 2},
+		"topic1": {0},
+		"topic2": {0, 1},
+		"topic3": {0, 1, 2},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1323,16 +1364,16 @@ func Test_stickyBalanceStrategy_Plan_KIP54_ExampleThree(t *testing.T) {
 
 	// PLAN 1
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{
+		"consumer1": {
 			Topics: topicNames,
 		},
-		"consumer2": ConsumerGroupMemberMetadata{
+		"consumer2": {
 			Topics: topicNames,
 		},
 	}
 	topics := map[string][]int32{
-		"topic1": []int32{0, 1},
-		"topic2": []int32{0, 1},
+		"topic1": {0, 1},
+		"topic2": {0, 1},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1359,12 +1400,12 @@ func Test_stickyBalanceStrategy_Plan_AddRemoveConsumerOneTopic(t *testing.T) {
 
 	// PLAN 1
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{
+		"consumer1": {
 			Topics: []string{"topic"},
 		},
 	}
 	topics := map[string][]int32{
-		"topic": []int32{0, 1, 2},
+		"topic": {0, 1, 2},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1395,16 +1436,16 @@ func Test_stickyBalanceStrategy_Plan_PoorRoundRobinAssignmentScenario(t *testing
 
 	// PLAN 1
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{
+		"consumer1": {
 			Topics: []string{"topic1", "topic2", "topic3", "topic4", "topic5"},
 		},
-		"consumer2": ConsumerGroupMemberMetadata{
+		"consumer2": {
 			Topics: []string{"topic1", "topic3", "topic5"},
 		},
-		"consumer3": ConsumerGroupMemberMetadata{
+		"consumer3": {
 			Topics: []string{"topic1", "topic3", "topic5"},
 		},
-		"consumer4": ConsumerGroupMemberMetadata{
+		"consumer4": {
 			Topics: []string{"topic1", "topic2", "topic3", "topic4", "topic5"},
 		},
 	}
@@ -1426,15 +1467,15 @@ func Test_stickyBalanceStrategy_Plan_AddRemoveTopicTwoConsumers(t *testing.T) {
 
 	// PLAN 1
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{
+		"consumer1": {
 			Topics: []string{"topic1"},
 		},
-		"consumer2": ConsumerGroupMemberMetadata{
+		"consumer2": {
 			Topics: []string{"topic1"},
 		},
 	}
 	topics := map[string][]int32{
-		"topic1": []int32{0, 1, 2},
+		"topic1": {0, 1, 2},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1700,15 +1741,15 @@ func Test_stickyBalanceStrategy_Plan_MoveExistingAssignments(t *testing.T) {
 	members := make(map[string]ConsumerGroupMemberMetadata, 3)
 	members["consumer1"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1", "topic2"},
-		UserData: encodeSubscriberPlan(t, map[string][]int32{"topic1": []int32{0}}),
+		UserData: encodeSubscriberPlan(t, map[string][]int32{"topic1": {0}}),
 	}
 	members["consumer2"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1", "topic2", "topic3", "topic4"},
-		UserData: encodeSubscriberPlan(t, map[string][]int32{"topic2": []int32{0}, "topic3": []int32{0}}),
+		UserData: encodeSubscriberPlan(t, map[string][]int32{"topic2": {0}, "topic3": {0}}),
 	}
 	members["consumer3"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic2", "topic3", "topic4", "topic5", "topic6"},
-		UserData: encodeSubscriberPlan(t, map[string][]int32{"topic4": []int32{0}, "topic5": []int32{0}, "topic6": []int32{0}}),
+		UserData: encodeSubscriberPlan(t, map[string][]int32{"topic4": {0}, "topic5": {0}, "topic6": {0}}),
 	}
 
 	plan, err := s.Plan(members, topics)
@@ -1718,12 +1759,12 @@ func Test_stickyBalanceStrategy_Plan_MoveExistingAssignments(t *testing.T) {
 func Test_stickyBalanceStrategy_Plan_Stickiness(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1, 2}}
+	topics := map[string][]int32{"topic1": {0, 1, 2}}
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer2": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer3": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer4": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
+		"consumer1": {Topics: []string{"topic1"}},
+		"consumer2": {Topics: []string{"topic1"}},
+		"consumer3": {Topics: []string{"topic1"}},
+		"consumer4": {Topics: []string{"topic1"}},
 	}
 
 	plan1, err := s.Plan(members, topics)
@@ -1753,7 +1794,7 @@ func Test_stickyBalanceStrategy_Plan_AssignmentUpdatedForDeletedTopic(t *testing
 		topics["topic3"][i] = int32(i)
 	}
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{Topics: []string{"topic1", "topic2", "topic3"}},
+		"consumer1": {Topics: []string{"topic1", "topic2", "topic3"}},
 	}
 
 	plan, err := s.Plan(members, topics)
@@ -1768,9 +1809,9 @@ func Test_stickyBalanceStrategy_Plan_AssignmentUpdatedForDeletedTopic(t *testing
 func Test_stickyBalanceStrategy_Plan_NoExceptionRaisedWhenOnlySubscribedTopicDeleted(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1, 2}}
+	topics := map[string][]int32{"topic1": {0, 1, 2}}
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
+		"consumer1": {Topics: []string{"topic1"}},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1796,11 +1837,11 @@ func Test_stickyBalanceStrategy_Plan_NoExceptionRaisedWhenOnlySubscribedTopicDel
 func Test_stickyBalanceStrategy_Plan_AssignmentWithMultipleGenerations1(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1, 2, 3, 4, 5}}
+	topics := map[string][]int32{"topic1": {0, 1, 2, 3, 4, 5}}
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer2": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer3": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
+		"consumer1": {Topics: []string{"topic1"}},
+		"consumer2": {Topics: []string{"topic1"}},
+		"consumer3": {Topics: []string{"topic1"}},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1846,11 +1887,11 @@ func Test_stickyBalanceStrategy_Plan_AssignmentWithMultipleGenerations1(t *testi
 func Test_stickyBalanceStrategy_Plan_AssignmentWithMultipleGenerations2(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1, 2, 3, 4, 5}}
+	topics := map[string][]int32{"topic1": {0, 1, 2, 3, 4, 5}}
 	members := map[string]ConsumerGroupMemberMetadata{
-		"consumer1": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer2": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
-		"consumer3": ConsumerGroupMemberMetadata{Topics: []string{"topic1"}},
+		"consumer1": {Topics: []string{"topic1"}},
+		"consumer2": {Topics: []string{"topic1"}},
+		"consumer3": {Topics: []string{"topic1"}},
 	}
 	plan1, err := s.Plan(members, topics)
 	verifyPlanIsBalancedAndSticky(t, s, members, plan1, err)
@@ -1891,19 +1932,19 @@ func Test_stickyBalanceStrategy_Plan_AssignmentWithMultipleGenerations2(t *testi
 func Test_stickyBalanceStrategy_Plan_AssignmentWithConflictingPreviousGenerations(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1, 2, 3, 4, 5}}
+	topics := map[string][]int32{"topic1": {0, 1, 2, 3, 4, 5}}
 	members := make(map[string]ConsumerGroupMemberMetadata, 3)
 	members["consumer1"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0, 1, 4}}, 1),
+		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0, 1, 4}}, 1),
 	}
 	members["consumer2"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0, 2, 3}}, 1),
+		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0, 2, 3}}, 1),
 	}
 	members["consumer3"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{3, 4, 5}}, 2),
+		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {3, 4, 5}}, 2),
 	}
 
 	plan, err := s.Plan(members, topics)
@@ -1914,15 +1955,15 @@ func Test_stickyBalanceStrategy_Plan_AssignmentWithConflictingPreviousGeneration
 func Test_stickyBalanceStrategy_Plan_SchemaBackwardCompatibility(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1, 2}}
+	topics := map[string][]int32{"topic1": {0, 1, 2}}
 	members := make(map[string]ConsumerGroupMemberMetadata, 3)
 	members["consumer1"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0, 2}}, 1),
+		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0, 2}}, 1),
 	}
 	members["consumer2"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithOldSchema(t, map[string][]int32{"topic1": []int32{1}}),
+		UserData: encodeSubscriberPlanWithOldSchema(t, map[string][]int32{"topic1": {1}}),
 	}
 	members["consumer3"] = ConsumerGroupMemberMetadata{Topics: []string{"topic1"}}
 
@@ -1934,15 +1975,15 @@ func Test_stickyBalanceStrategy_Plan_SchemaBackwardCompatibility(t *testing.T) {
 func Test_stickyBalanceStrategy_Plan_ConflictingPreviousAssignments(t *testing.T) {
 	s := &stickyBalanceStrategy{}
 
-	topics := map[string][]int32{"topic1": []int32{0, 1}}
+	topics := map[string][]int32{"topic1": {0, 1}}
 	members := make(map[string]ConsumerGroupMemberMetadata, 2)
 	members["consumer1"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0, 1}}, 1),
+		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0, 1}}, 1),
 	}
 	members["consumer2"] = ConsumerGroupMemberMetadata{
 		Topics:   []string{"topic1"},
-		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": []int32{0, 1}}, 1),
+		UserData: encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0, 1}}, 1),
 	}
 
 	plan, err := s.Plan(members, topics)
@@ -1950,6 +1991,28 @@ func Test_stickyBalanceStrategy_Plan_ConflictingPreviousAssignments(t *testing.T
 	verifyFullyBalanced(t, plan)
 }
 
+func Test_stickyBalanceStrategy_Plan_AssignmentData(t *testing.T) {
+	s := &stickyBalanceStrategy{}
+
+	members := make(map[string]ConsumerGroupMemberMetadata, 2)
+	members["consumer1"] = ConsumerGroupMemberMetadata{
+		Topics: []string{"topic1"},
+	}
+	members["consumer2"] = ConsumerGroupMemberMetadata{
+		Topics: []string{"topic1"},
+	}
+
+	expected := encodeSubscriberPlanWithGeneration(t, map[string][]int32{"topic1": {0, 1}}, 1)
+
+	actual, err := s.AssignmentData("consumer1", map[string][]int32{"topic1": {0, 1}}, 1)
+	if err != nil {
+		t.Errorf("Error building assignment data: %v", err)
+	}
+	if !bytes.Equal(expected, actual) {
+		t.Error("Invalid assignment data returned from AssignmentData")
+	}
+}
+
 func BenchmarkStickAssignmentWithLargeNumberOfConsumersAndTopics(b *testing.B) {
 	s := &stickyBalanceStrategy{}
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -2228,14 +2291,14 @@ func Test_sortPartitionsByPotentialConsumerAssignments(t *testing.T) {
 			name: "Single topic partition",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{
+					{
 						Topic:     "t1",
 						Partition: 0,
-					}: []string{"c1", "c2"},
+					}: {"c1", "c2"},
 				},
 			},
 			want: []topicPartitionAssignment{
-				topicPartitionAssignment{
+				{
 					Topic:     "t1",
 					Partition: 0,
 				},
@@ -2245,22 +2308,22 @@ func Test_sortPartitionsByPotentialConsumerAssignments(t *testing.T) {
 			name: "Multiple topic partitions with the same number of consumers but different topic names",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{
+					{
 						Topic:     "t1",
 						Partition: 0,
-					}: []string{"c1", "c2"},
-					topicPartitionAssignment{
+					}: {"c1", "c2"},
+					{
 						Topic:     "t2",
 						Partition: 0,
-					}: []string{"c1", "c2"},
+					}: {"c1", "c2"},
 				},
 			},
 			want: []topicPartitionAssignment{
-				topicPartitionAssignment{
+				{
 					Topic:     "t1",
 					Partition: 0,
 				},
-				topicPartitionAssignment{
+				{
 					Topic:     "t2",
 					Partition: 0,
 				},
@@ -2270,22 +2333,22 @@ func Test_sortPartitionsByPotentialConsumerAssignments(t *testing.T) {
 			name: "Multiple topic partitions with the same number of consumers and topic names",
 			args: args{
 				partition2AllPotentialConsumers: map[topicPartitionAssignment][]string{
-					topicPartitionAssignment{
+					{
 						Topic:     "t1",
 						Partition: 0,
-					}: []string{"c1", "c2"},
-					topicPartitionAssignment{
+					}: {"c1", "c2"},
+					{
 						Topic:     "t1",
 						Partition: 1,
-					}: []string{"c1", "c2"},
+					}: {"c1", "c2"},
 				},
 			},
 			want: []topicPartitionAssignment{
-				topicPartitionAssignment{
+				{
 					Topic:     "t1",
 					Partition: 0,
 				},
-				topicPartitionAssignment{
+				{
 					Topic:     "t1",
 					Partition: 1,
 				},
diff --git a/broker_test.go b/broker_test.go
index fe170ab..387dae9 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"net"
@@ -108,7 +109,6 @@ func TestSimpleBrokerCommunication(t *testing.T) {
 			t.Error(err)
 		}
 	}
-
 }
 
 var ErrTokenFailure = errors.New("Failure generating token")
@@ -130,9 +130,9 @@ func newTokenProvider(token *AccessToken, err error) *TokenProvider {
 }
 
 func TestSASLOAuthBearer(t *testing.T) {
-
 	testTable := []struct {
 		name                      string
+		authidentity              string
 		mockSASLHandshakeResponse MockResponse // Mock SaslHandshakeRequest response from broker
 		mockSASLAuthResponse      MockResponse // Mock SaslAuthenticateRequest response from broker
 		expectClientErr           bool         // Expect an internal client-side error
@@ -200,7 +200,6 @@ func TestSASLOAuthBearer(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-
 		// mockBroker mocks underlying network logic and broker responses
 		mockBroker := NewMockBroker(t, 0)
 
@@ -218,6 +217,7 @@ func TestSASLOAuthBearer(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypeOAuth
@@ -245,10 +245,6 @@ func TestSASLOAuthBearer(t *testing.T) {
 			if test.expectedBrokerError != err {
 				t.Errorf("[%d]:[%s] Expected %s auth error, got %s\n", i, test.name, test.expectedBrokerError, err)
 			}
-		} else if test.expectedBrokerError != ErrNoError {
-			if test.expectedBrokerError != err {
-				t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.expectedBrokerError, err)
-			}
 		} else if test.expectClientErr && err == nil {
 			t.Errorf("[%d]:[%s] Expected a client error and got none\n", i, test.name)
 		} else if !test.expectClientErr && err != nil {
@@ -325,7 +321,6 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-
 		// mockBroker mocks underlying network logic and broker responses
 		mockBroker := NewMockBroker(t, 0)
 		broker := NewBroker(mockBroker.Addr())
@@ -337,6 +332,7 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 
 		mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).SetAuthBytes([]byte(test.scramChallengeResp))
 		mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).SetEnabledMechanisms([]string{SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512})
@@ -393,9 +389,9 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 }
 
 func TestSASLPlainAuth(t *testing.T) {
-
 	testTable := []struct {
 		name             string
+		authidentity     string
 		mockAuthErr      KError // Mock and expect error returned from SaslAuthenticateRequest
 		mockHandshakeErr KError // Mock and expect error returned from SaslHandshakeRequest
 		expectClientErr  bool   // Expect an internal client-side error
@@ -405,6 +401,12 @@ func TestSASLPlainAuth(t *testing.T) {
 			mockAuthErr:      ErrNoError,
 			mockHandshakeErr: ErrNoError,
 		},
+		{
+			name:             "SASL Plain OK server response with authidentity",
+			authidentity:     "authid",
+			mockAuthErr:      ErrNoError,
+			mockHandshakeErr: ErrNoError,
+		},
 		{
 			name:             "SASL Plain authentication failure response",
 			mockAuthErr:      ErrSASLAuthenticationFailed,
@@ -418,7 +420,6 @@ func TestSASLPlainAuth(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-
 		// mockBroker mocks underlying network logic and broker responses
 		mockBroker := NewMockBroker(t, 0)
 
@@ -450,9 +451,11 @@ func TestSASLPlainAuth(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypePlaintext
+		conf.Net.SASL.AuthIdentity = test.authidentity
 		conf.Net.SASL.User = "token"
 		conf.Net.SASL.Password = "password"
 		conf.Net.SASL.Version = SASLHandshakeV1
@@ -474,6 +477,23 @@ func TestSASLPlainAuth(t *testing.T) {
 		broker.conn = conn
 
 		err = broker.authenticateViaSASL()
+		if err == nil {
+			for _, rr := range mockBroker.History() {
+				switch r := rr.Request.(type) {
+				case *SaslAuthenticateRequest:
+					x := bytes.SplitN(r.SaslAuthBytes, []byte("\x00"), 3)
+					if string(x[0]) != conf.Net.SASL.AuthIdentity {
+						t.Errorf("[%d]:[%s] expected %s auth identity, got %s\n", i, test.name, conf.Net.SASL.AuthIdentity, x[0])
+					}
+					if string(x[1]) != conf.Net.SASL.User {
+						t.Errorf("[%d]:[%s] expected %s user, got %s\n", i, test.name, conf.Net.SASL.User, x[1])
+					}
+					if string(x[2]) != conf.Net.SASL.Password {
+						t.Errorf("[%d]:[%s] expected %s password, got %s\n", i, test.name, conf.Net.SASL.Password, x[2])
+					}
+				}
+			}
+		}
 
 		if test.mockAuthErr != ErrNoError {
 			if test.mockAuthErr != err {
@@ -515,6 +535,7 @@ func TestSASLReadTimeout(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 	}
 
 	conf := NewConfig()
@@ -543,7 +564,6 @@ func TestSASLReadTimeout(t *testing.T) {
 }
 
 func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
-
 	testTable := []struct {
 		name               string
 		error              error
@@ -606,6 +626,7 @@ func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypeGSSAPI
 		conf.Net.SASL.GSSAPI.ServiceName = "kafka"
@@ -659,11 +680,9 @@ func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
 
 		mockBroker.Close()
 	}
-
 }
 
 func TestBuildClientFirstMessage(t *testing.T) {
-
 	testTable := []struct {
 		name        string
 		token       *AccessToken
@@ -700,7 +719,6 @@ func TestBuildClientFirstMessage(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-
 		actual, err := buildClientFirstMessage(test.token)
 
 		if !reflect.DeepEqual(test.expected, actual) {
@@ -973,6 +991,9 @@ func validateBrokerMetrics(t *testing.T, broker *Broker, mockBrokerMetrics broke
 	metricValidators.registerForAllBrokers(broker, countHistogramValidator("request-size", 1))
 	metricValidators.registerForAllBrokers(broker, minMaxHistogramValidator("request-size", mockBrokerBytesRead, mockBrokerBytesRead))
 
+	// Check that there is no more requests in flight
+	metricValidators.registerForAllBrokers(broker, counterValidator("requests-in-flight", 0))
+
 	// Run the validators
 	metricValidators.run(t, broker.conf.MetricRegistry)
 }
diff --git a/client_test.go b/client_test.go
index 86fecac..2beef6c 100644
--- a/client_test.go
+++ b/client_test.go
@@ -830,13 +830,11 @@ func TestClientAutorefreshShutdownRace(t *testing.T) {
 	// Wait for the background refresh to kick in
 	time.Sleep(110 * time.Millisecond)
 
-	done := make(chan none)
+	errCh := make(chan error, 1)
 	go func() {
 		// Close the client
-		if err := client.Close(); err != nil {
-			t.Fatal(err)
-		}
-		close(done)
+		errCh <- client.Close()
+		close(errCh)
 	}()
 
 	// Wait for the Close to kick in
@@ -848,7 +846,10 @@ func TestClientAutorefreshShutdownRace(t *testing.T) {
 	metadataResponse.AddTopicPartition("foo", 0, leader.BrokerID(), []int32{2}, []int32{2}, []int32{}, ErrNoError)
 	seedBroker.Returns(metadataResponse)
 
-	<-done
+	err = <-errCh
+	if err != nil {
+		t.Fatalf("goroutine client.Close():%s", err)
+	}
 
 	seedBroker.Close()
 
diff --git a/client_tls_test.go b/client_tls_test.go
index eff84c7..0e47e17 100644
--- a/client_tls_test.go
+++ b/client_tls_test.go
@@ -86,7 +86,7 @@ func TestTLS(t *testing.T) {
 
 	// Keep server the same - it's the client that we're testing
 	serverTLSConfig := &tls.Config{
-		Certificates: []tls.Certificate{tls.Certificate{
+		Certificates: []tls.Certificate{{
 			Certificate: [][]byte{hostDer},
 			PrivateKey:  hostkey,
 		}},
@@ -103,7 +103,7 @@ func TestTLS(t *testing.T) {
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
 				RootCAs: systemCerts,
-				Certificates: []tls.Certificate{tls.Certificate{
+				Certificates: []tls.Certificate{{
 					Certificate: [][]byte{clientDer},
 					PrivateKey:  clientkey,
 				}},
@@ -114,7 +114,7 @@ func TestTLS(t *testing.T) {
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
 				RootCAs: pool,
-				Certificates: []tls.Certificate{tls.Certificate{
+				Certificates: []tls.Certificate{{
 					Certificate: [][]byte{clientDer},
 					PrivateKey:  hostkey,
 				}},
@@ -125,7 +125,7 @@ func TestTLS(t *testing.T) {
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
 				RootCAs: pool,
-				Certificates: []tls.Certificate{tls.Certificate{
+				Certificates: []tls.Certificate{{
 					Certificate: [][]byte{hostDer},
 					PrivateKey:  clientkey,
 				}},
@@ -135,7 +135,7 @@ func TestTLS(t *testing.T) {
 			Succeed: false,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
-				Certificates: []tls.Certificate{tls.Certificate{
+				Certificates: []tls.Certificate{{
 					Certificate: [][]byte{clientDer},
 					PrivateKey:  clientkey,
 				}},
@@ -153,7 +153,7 @@ func TestTLS(t *testing.T) {
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
 				RootCAs: pool,
-				Certificates: []tls.Certificate{tls.Certificate{
+				Certificates: []tls.Certificate{{
 					Certificate: [][]byte{clientDer},
 					PrivateKey:  clientkey,
 				}},
diff --git a/config_test.go b/config_test.go
index 3f7a7a7..a7267b1 100644
--- a/config_test.go
+++ b/config_test.go
@@ -194,7 +194,6 @@ func TestNetConfigValidates(t *testing.T) {
 				cfg.Net.SASL.GSSAPI.Username = "sarama"
 				cfg.Net.SASL.GSSAPI.Password = "sarama"
 				cfg.Net.SASL.GSSAPI.KerberosConfigPath = "/etc/krb5.conf"
-
 			},
 			"Net.SASL.GSSAPI.Realm must not be empty when GSS-API mechanism is used"},
 	}
@@ -405,6 +404,18 @@ func TestLZ4ConfigValidation(t *testing.T) {
 	}
 }
 
+func TestZstdConfigValidation(t *testing.T) {
+	config := NewConfig()
+	config.Producer.Compression = CompressionZSTD
+	if err := config.Validate(); string(err.(ConfigurationError)) != "zstd compression requires Version >= V2_1_0_0" {
+		t.Error("Expected invalid zstd/kafka version error, got ", err)
+	}
+	config.Version = V2_1_0_0
+	if err := config.Validate(); err != nil {
+		t.Error("Expected zstd to work, got ", err)
+	}
+}
+
 // This example shows how to integrate with an existing registry as well as publishing metrics
 // on the standard output
 func ExampleConfig_metrics() {
diff --git a/consumer_group_test.go b/consumer_group_test.go
index 8bf44e6..bbef974 100644
--- a/consumer_group_test.go
+++ b/consumer_group_test.go
@@ -18,20 +18,11 @@ func (h exampleConsumerGroupHandler) ConsumeClaim(sess ConsumerGroupSession, cla
 }
 
 func ExampleConsumerGroup() {
-	// Init config, specify appropriate version
 	config := NewConfig()
-	config.Version = V1_0_0_0
+	config.Version = V2_0_0_0 // specify appropriate version
 	config.Consumer.Return.Errors = true
 
-	// Start with a client
-	client, err := NewClient([]string{"localhost:9092"}, config)
-	if err != nil {
-		panic(err)
-	}
-	defer func() { _ = client.Close() }()
-
-	// Start a new consumer group
-	group, err := NewConsumerGroupFromClient("my-group", client)
+	group, err := NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)
 	if err != nil {
 		panic(err)
 	}
@@ -50,6 +41,9 @@ func ExampleConsumerGroup() {
 		topics := []string{"my-topic"}
 		handler := exampleConsumerGroupHandler{}
 
+		// `Consume` should be called inside an infinite loop, when a
+		// server-side rebalance happens, the consumer session will need to be
+		// recreated to get the new claims
 		err := group.Consume(ctx, topics, handler)
 		if err != nil {
 			panic(err)
diff --git a/consumer_test.go b/consumer_test.go
index 4107dfa..7b29ebc 100644
--- a/consumer_test.go
+++ b/consumer_test.go
@@ -1016,7 +1016,6 @@ func TestConsumerTimestamps(t *testing.T) {
 	now := time.Now().Truncate(time.Millisecond)
 	type testMessage struct {
 		key       Encoder
-		value     Encoder
 		offset    int64
 		timestamp time.Time
 	}
@@ -1027,32 +1026,32 @@ func TestConsumerTimestamps(t *testing.T) {
 		expectedTimestamp []time.Time
 	}{
 		{MinVersion, false, []testMessage{
-			{nil, testMsg, 1, now},
-			{nil, testMsg, 2, now},
+			{testMsg, 1, now},
+			{testMsg, 2, now},
 		}, []time.Time{{}, {}}},
 		{V0_9_0_0, false, []testMessage{
-			{nil, testMsg, 1, now},
-			{nil, testMsg, 2, now},
+			{testMsg, 1, now},
+			{testMsg, 2, now},
 		}, []time.Time{{}, {}}},
 		{V0_10_0_0, false, []testMessage{
-			{nil, testMsg, 1, now},
-			{nil, testMsg, 2, now},
+			{testMsg, 1, now},
+			{testMsg, 2, now},
 		}, []time.Time{{}, {}}},
 		{V0_10_2_1, false, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now.Add(time.Second), now.Add(2 * time.Second)}},
 		{V0_10_2_1, true, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now, now}},
 		{V0_11_0_0, false, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now.Add(time.Second), now.Add(2 * time.Second)}},
 		{V0_11_0_0, true, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now, now}},
 	} {
 		var fr *FetchResponse
diff --git a/create_partitions_request_test.go b/create_partitions_request_test.go
index db8cb73..8230b38 100644
--- a/create_partitions_request_test.go
+++ b/create_partitions_request_test.go
@@ -32,7 +32,7 @@ var (
 func TestCreatePartitionsRequest(t *testing.T) {
 	req := &CreatePartitionsRequest{
 		TopicPartitions: map[string]*TopicPartition{
-			"topic": &TopicPartition{
+			"topic": {
 				Count: 3,
 			},
 		},
diff --git a/create_partitions_response_test.go b/create_partitions_response_test.go
index 815aba1..2f59674 100644
--- a/create_partitions_response_test.go
+++ b/create_partitions_response_test.go
@@ -28,7 +28,7 @@ func TestCreatePartitionsResponse(t *testing.T) {
 	resp := &CreatePartitionsResponse{
 		ThrottleTime: 100 * time.Millisecond,
 		TopicPartitionErrors: map[string]*TopicPartitionError{
-			"topic": &TopicPartitionError{},
+			"topic": {},
 		},
 	}
 
diff --git a/create_topics_request_test.go b/create_topics_request_test.go
index 56b1b80..daf223c 100644
--- a/create_topics_request_test.go
+++ b/create_topics_request_test.go
@@ -31,7 +31,7 @@ func TestCreateTopicsRequest(t *testing.T) {
 				NumPartitions:     -1,
 				ReplicationFactor: -1,
 				ReplicaAssignment: map[int32][]int32{
-					0: []int32{0, 1, 2},
+					0: {0, 1, 2},
 				},
 				ConfigEntries: map[string]*string{
 					"retention.ms": &retention,
diff --git a/create_topics_response_test.go b/create_topics_response_test.go
index 30c1fb2..2dfb8d1 100644
--- a/create_topics_response_test.go
+++ b/create_topics_response_test.go
@@ -31,7 +31,7 @@ var (
 func TestCreateTopicsResponse(t *testing.T) {
 	resp := &CreateTopicsResponse{
 		TopicErrors: map[string]*TopicError{
-			"topic": &TopicError{
+			"topic": {
 				Err: ErrInvalidRequest,
 			},
 		},
diff --git a/describe_configs_request_test.go b/describe_configs_request_test.go
index fec2f11..ef9d3ff 100644
--- a/describe_configs_request_test.go
+++ b/describe_configs_request_test.go
@@ -61,7 +61,7 @@ func TestDescribeConfigsRequestv0(t *testing.T) {
 	request = &DescribeConfigsRequest{
 		Version: 0,
 		Resources: []*ConfigResource{
-			&ConfigResource{
+			{
 				Type:        TopicResource,
 				Name:        "foo",
 				ConfigNames: configs,
@@ -74,12 +74,12 @@ func TestDescribeConfigsRequestv0(t *testing.T) {
 	request = &DescribeConfigsRequest{
 		Version: 0,
 		Resources: []*ConfigResource{
-			&ConfigResource{
+			{
 				Type:        TopicResource,
 				Name:        "foo",
 				ConfigNames: []string{"segment.ms", "retention.ms"},
 			},
-			&ConfigResource{
+			{
 				Type:        TopicResource,
 				Name:        "bar",
 				ConfigNames: []string{"segment.ms"},
@@ -91,7 +91,7 @@ func TestDescribeConfigsRequestv0(t *testing.T) {
 	request = &DescribeConfigsRequest{
 		Version: 0,
 		Resources: []*ConfigResource{
-			&ConfigResource{
+			{
 				Type: TopicResource,
 				Name: "foo",
 			},
@@ -102,9 +102,7 @@ func TestDescribeConfigsRequestv0(t *testing.T) {
 }
 
 func TestDescribeConfigsRequestv1(t *testing.T) {
-	var request *DescribeConfigsRequest
-
-	request = &DescribeConfigsRequest{
+	request := &DescribeConfigsRequest{
 		Version: 1,
 		Resources: []*ConfigResource{
 			{
diff --git a/describe_configs_response_test.go b/describe_configs_response_test.go
index 9584d57..c91a042 100644
--- a/describe_configs_response_test.go
+++ b/describe_configs_response_test.go
@@ -25,6 +25,21 @@ var (
 		0, // Sensitive
 	}
 
+	describeConfigsResponseWithDefaultv0 = []byte{
+		0, 0, 0, 0, //throttle
+		0, 0, 0, 1, // response
+		0, 0, //errorcode
+		0, 0, //string
+		2, // topic
+		0, 3, 'f', 'o', 'o',
+		0, 0, 0, 1, //configs
+		0, 10, 's', 'e', 'g', 'm', 'e', 'n', 't', '.', 'm', 's',
+		0, 4, '1', '0', '0', '0',
+		0, // ReadOnly
+		1, // Default
+		0, // Sensitive
+	}
+
 	describeConfigsResponsePopulatedv1 = []byte{
 		0, 0, 0, 0, //throttle
 		0, 0, 0, 1, // response
@@ -59,6 +74,22 @@ var (
 		0, 4, '1', '0', '0', '0',
 		4, // Source
 	}
+
+	describeConfigsResponseWithDefaultv1 = []byte{
+		0, 0, 0, 0, //throttle
+		0, 0, 0, 1, // response
+		0, 0, //errorcode
+		0, 0, //string
+		2, // topic
+		0, 3, 'f', 'o', 'o',
+		0, 0, 0, 1, //configs
+		0, 10, 's', 'e', 'g', 'm', 'e', 'n', 't', '.', 'm', 's',
+		0, 4, '1', '0', '0', '0',
+		0,          // ReadOnly
+		5,          // Source
+		0,          // Sensitive
+		0, 0, 0, 0, // No Synonym
+	}
 )
 
 func TestDescribeConfigsResponsev0(t *testing.T) {
@@ -74,18 +105,19 @@ func TestDescribeConfigsResponsev0(t *testing.T) {
 
 	response = &DescribeConfigsResponse{
 		Version: 0, Resources: []*ResourceResponse{
-			&ResourceResponse{
+			{
 				ErrorCode: 0,
 				ErrorMsg:  "",
 				Type:      TopicResource,
 				Name:      "foo",
 				Configs: []*ConfigEntry{
-					&ConfigEntry{
+					{
 						Name:      "segment.ms",
 						Value:     "1000",
 						ReadOnly:  false,
 						Default:   false,
 						Sensitive: false,
+						Source:    SourceUnknown,
 					},
 				},
 			},
@@ -94,6 +126,40 @@ func TestDescribeConfigsResponsev0(t *testing.T) {
 	testResponse(t, "response with error", response, describeConfigsResponsePopulatedv0)
 }
 
+func TestDescribeConfigsResponseWithDefaultv0(t *testing.T) {
+	var response *DescribeConfigsResponse
+
+	response = &DescribeConfigsResponse{
+		Resources: []*ResourceResponse{},
+	}
+	testVersionDecodable(t, "empty", response, describeConfigsResponseEmpty, 0)
+	if len(response.Resources) != 0 {
+		t.Error("Expected no groups")
+	}
+
+	response = &DescribeConfigsResponse{
+		Version: 0, Resources: []*ResourceResponse{
+			{
+				ErrorCode: 0,
+				ErrorMsg:  "",
+				Type:      TopicResource,
+				Name:      "foo",
+				Configs: []*ConfigEntry{
+					{
+						Name:      "segment.ms",
+						Value:     "1000",
+						ReadOnly:  false,
+						Default:   true,
+						Sensitive: false,
+						Source:    SourceDefault,
+					},
+				},
+			},
+		},
+	}
+	testResponse(t, "response with default", response, describeConfigsResponseWithDefaultv0)
+}
+
 func TestDescribeConfigsResponsev1(t *testing.T) {
 	var response *DescribeConfigsResponse
 
@@ -108,17 +174,18 @@ func TestDescribeConfigsResponsev1(t *testing.T) {
 	response = &DescribeConfigsResponse{
 		Version: 1,
 		Resources: []*ResourceResponse{
-			&ResourceResponse{
+			{
 				ErrorCode: 0,
 				ErrorMsg:  "",
 				Type:      TopicResource,
 				Name:      "foo",
 				Configs: []*ConfigEntry{
-					&ConfigEntry{
+					{
 						Name:      "segment.ms",
 						Value:     "1000",
 						ReadOnly:  false,
 						Source:    SourceStaticBroker,
+						Default:   false,
 						Sensitive: false,
 						Synonyms:  []*ConfigSynonym{},
 					},
@@ -143,17 +210,18 @@ func TestDescribeConfigsResponseWithSynonym(t *testing.T) {
 	response = &DescribeConfigsResponse{
 		Version: 1,
 		Resources: []*ResourceResponse{
-			&ResourceResponse{
+			{
 				ErrorCode: 0,
 				ErrorMsg:  "",
 				Type:      TopicResource,
 				Name:      "foo",
 				Configs: []*ConfigEntry{
-					&ConfigEntry{
+					{
 						Name:      "segment.ms",
 						Value:     "1000",
 						ReadOnly:  false,
 						Source:    SourceStaticBroker,
+						Default:   false,
 						Sensitive: false,
 						Synonyms: []*ConfigSynonym{
 							{
@@ -169,3 +237,39 @@ func TestDescribeConfigsResponseWithSynonym(t *testing.T) {
 	}
 	testResponse(t, "response with error", response, describeConfigsResponseWithSynonymv1)
 }
+
+func TestDescribeConfigsResponseWithDefaultv1(t *testing.T) {
+	var response *DescribeConfigsResponse
+
+	response = &DescribeConfigsResponse{
+		Resources: []*ResourceResponse{},
+	}
+	testVersionDecodable(t, "empty", response, describeConfigsResponseEmpty, 0)
+	if len(response.Resources) != 0 {
+		t.Error("Expected no groups")
+	}
+
+	response = &DescribeConfigsResponse{
+		Version: 1,
+		Resources: []*ResourceResponse{
+			{
+				ErrorCode: 0,
+				ErrorMsg:  "",
+				Type:      TopicResource,
+				Name:      "foo",
+				Configs: []*ConfigEntry{
+					{
+						Name:      "segment.ms",
+						Value:     "1000",
+						ReadOnly:  false,
+						Source:    SourceDefault,
+						Default:   true,
+						Sensitive: false,
+						Synonyms:  []*ConfigSynonym{},
+					},
+				},
+			},
+		},
+	}
+	testResponse(t, "response with error", response, describeConfigsResponseWithDefaultv1)
+}
diff --git a/describe_log_dirs_request_test.go b/describe_log_dirs_request_test.go
index 9846a6c..65b88c6 100644
--- a/describe_log_dirs_request_test.go
+++ b/describe_log_dirs_request_test.go
@@ -22,7 +22,7 @@ func TestDescribeLogDirsRequest(t *testing.T) {
 	testRequest(t, "no topics", request, emptyDescribeLogDirsRequest)
 
 	request.DescribeTopics = []DescribeLogDirsRequestTopic{
-		DescribeLogDirsRequestTopic{
+		{
 			Topic:        "random",
 			PartitionIDs: []int32{25, 26},
 		},
diff --git a/describe_log_dirs_response_test.go b/describe_log_dirs_response_test.go
index 3bffafc..5a1eb5b 100644
--- a/describe_log_dirs_response_test.go
+++ b/describe_log_dirs_response_test.go
@@ -42,20 +42,20 @@ func TestDescribeLogDirsResponse(t *testing.T) {
 	}
 
 	response.LogDirs = []DescribeLogDirsResponseDirMetadata{
-		DescribeLogDirsResponseDirMetadata{
+		{
 			ErrorCode: 0,
 			Path:      "/kafka",
 			Topics: []DescribeLogDirsResponseTopic{
-				DescribeLogDirsResponseTopic{
+				{
 					Topic: "random",
 					Partitions: []DescribeLogDirsResponsePartition{
-						DescribeLogDirsResponsePartition{
+						{
 							PartitionID: 25,
 							Size:        125,
 							OffsetLag:   0,
 							IsTemporary: false,
 						},
-						DescribeLogDirsResponsePartition{
+						{
 							PartitionID: 26,
 							Size:        100,
 							OffsetLag:   0,
diff --git a/fetch_request_test.go b/fetch_request_test.go
index 1a94c2d..0903caf 100644
--- a/fetch_request_test.go
+++ b/fetch_request_test.go
@@ -26,23 +26,65 @@ var (
 		0x00, 0x05, 't', 'o', 'p', 'i', 'c',
 		0x00, 0x00, 0x00, 0x01,
 		0x00, 0x00, 0x00, 0x12, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x34, 0x00, 0x00, 0x00, 0x56}
+
+	fetchRequestOneBlockV11 = []byte{
+		0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0xFF,
+		0x01,
+		0x00, 0x00, 0x00, 0xAA, // sessionID
+		0x00, 0x00, 0x00, 0xEE, // sessionEpoch
+		0x00, 0x00, 0x00, 0x01,
+		0x00, 0x05, 't', 'o', 'p', 'i', 'c',
+		0x00, 0x00, 0x00, 0x01,
+		0x00, 0x00, 0x00, 0x12, // partitionID
+		0xFF, 0xFF, 0xFF, 0xFF, // currentLeaderEpoch
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x34, // fetchOffset
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // logStartOffset
+		0x00, 0x00, 0x00, 0x56, // maxBytes
+		0x00, 0x00, 0x00, 0x00,
+		0x00, 0x06, 'r', 'a', 'c', 'k', '0', '1', // rackID
+	}
 )
 
 func TestFetchRequest(t *testing.T) {
-	request := new(FetchRequest)
-	testRequest(t, "no blocks", request, fetchRequestNoBlocks)
-
-	request.MaxWaitTime = 0x20
-	request.MinBytes = 0xEF
-	testRequest(t, "with properties", request, fetchRequestWithProperties)
-
-	request.MaxWaitTime = 0
-	request.MinBytes = 0
-	request.AddBlock("topic", 0x12, 0x34, 0x56)
-	testRequest(t, "one block", request, fetchRequestOneBlock)
-
-	request.Version = 4
-	request.MaxBytes = 0xFF
-	request.Isolation = ReadCommitted
-	testRequest(t, "one block v4", request, fetchRequestOneBlockV4)
+	t.Run("no blocks", func(t *testing.T) {
+		request := new(FetchRequest)
+		testRequest(t, "no blocks", request, fetchRequestNoBlocks)
+	})
+
+	t.Run("with properties", func(t *testing.T) {
+		request := new(FetchRequest)
+		request.MaxWaitTime = 0x20
+		request.MinBytes = 0xEF
+		testRequest(t, "with properties", request, fetchRequestWithProperties)
+	})
+
+	t.Run("one block", func(t *testing.T) {
+		request := new(FetchRequest)
+		request.MaxWaitTime = 0
+		request.MinBytes = 0
+		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		testRequest(t, "one block", request, fetchRequestOneBlock)
+	})
+
+	t.Run("one block v4", func(t *testing.T) {
+		request := new(FetchRequest)
+		request.Version = 4
+		request.MaxBytes = 0xFF
+		request.Isolation = ReadCommitted
+		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		testRequest(t, "one block v4", request, fetchRequestOneBlockV4)
+	})
+
+	t.Run("one block v11 rackid", 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.RackID = "rack01"
+		testRequest(t, "one block v11 rackid", request, fetchRequestOneBlockV11)
+	})
 }
diff --git a/fetch_response_test.go b/fetch_response_test.go
index 58a0987..e471581 100644
--- a/fetch_response_test.go
+++ b/fetch_response_test.go
@@ -137,6 +137,31 @@ var (
 		0x00,
 		0xFF, 0xFF, 0xFF, 0xFF,
 		0x00, 0x00, 0x00, 0x02, 0x00, 0xEE}
+
+	preferredReplicaFetchResponseV11 = []byte{
+		0x00, 0x00, 0x00, 0x00, // ThrottleTime
+		0x00, 0x02, // ErrorCode
+		0x00, 0x00, 0x00, 0xAC, // SessionID
+		0x00, 0x00, 0x00, 0x01, // Number of Topics
+		0x00, 0x05, 't', 'o', 'p', 'i', 'c', // Topic
+		0x00, 0x00, 0x00, 0x01, // Number of Partitions
+		0x00, 0x00, 0x00, 0x05, // Partition
+		0x00, 0x01, // Error
+		0x00, 0x00, 0x00, 0x00, 0x10, 0x10, 0x10, 0x10, // High Watermark Offset
+		0x00, 0x00, 0x00, 0x00, 0x10, 0x10, 0x10, 0x09, // Last Stable Offset
+		0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x01, 0x01, // Log Start Offset
+		0x00, 0x00, 0x00, 0x00, // Number of Aborted Transactions
+		0x00, 0x00, 0x00, 0x03, // Preferred Read Replica
+		0x00, 0x00, 0x00, 0x1C,
+		// messageSet
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x55, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x10,
+		// message
+		0x23, 0x96, 0x4a, 0xf7, // CRC
+		0x00,
+		0x00,
+		0xFF, 0xFF, 0xFF, 0xFF,
+		0x00, 0x00, 0x00, 0x02, 0x00, 0xEE}
 )
 
 func TestEmptyFetchResponse(t *testing.T) {
@@ -146,7 +171,6 @@ func TestEmptyFetchResponse(t *testing.T) {
 	if len(response.Blocks) != 0 {
 		t.Error("Decoding produced topic blocks where there were none.")
 	}
-
 }
 
 func TestOneMessageFetchResponse(t *testing.T) {
@@ -399,3 +423,75 @@ func TestOneMessageFetchResponseV4(t *testing.T) {
 		t.Error("Decoding produced incorrect message value.")
 	}
 }
+
+func TestPreferredReplicaFetchResponseV11(t *testing.T) {
+	response := FetchResponse{}
+	testVersionDecodable(
+		t, "preferred replica fetch response v11", &response,
+		preferredReplicaFetchResponseV11, 11)
+
+	if response.ErrorCode != 0x0002 {
+		t.Fatal("Decoding produced incorrect error code.")
+	}
+
+	if response.SessionID != 0x000000AC {
+		t.Fatal("Decoding produced incorrect session ID.")
+	}
+
+	if len(response.Blocks) != 1 {
+		t.Fatal("Decoding produced incorrect number of topic blocks.")
+	}
+
+	if len(response.Blocks["topic"]) != 1 {
+		t.Fatal("Decoding produced incorrect number of partition blocks for topic.")
+	}
+
+	block := response.GetBlock("topic", 5)
+	if block == nil {
+		t.Fatal("GetBlock didn't return block.")
+	}
+	if block.Err != ErrOffsetOutOfRange {
+		t.Error("Decoding didn't produce correct error code.")
+	}
+	if block.HighWaterMarkOffset != 0x10101010 {
+		t.Error("Decoding didn't produce correct high water mark offset.")
+	}
+	if block.LastStableOffset != 0x10101009 {
+		t.Error("Decoding didn't produce correct last stable offset.")
+	}
+	if block.LogStartOffset != 0x01010101 {
+		t.Error("Decoding didn't produce correct log start offset.")
+	}
+	if block.PreferredReadReplica != 0x0003 {
+		t.Error("Decoding didn't produce correct preferred read replica.")
+	}
+	partial, err := block.isPartial()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if partial {
+		t.Error("Decoding detected a partial trailing record where there wasn't one.")
+	}
+
+	n, err := block.numRecords()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if n != 1 {
+		t.Fatal("Decoding produced incorrect number of records.")
+	}
+	msgBlock := block.RecordsSet[0].MsgSet.Messages[0]
+	if msgBlock.Offset != 0x550000 {
+		t.Error("Decoding produced incorrect message offset.")
+	}
+	msg := msgBlock.Msg
+	if msg.Codec != CompressionNone {
+		t.Error("Decoding produced incorrect message compression.")
+	}
+	if msg.Key != nil {
+		t.Error("Decoding produced message key where there was none.")
+	}
+	if !bytes.Equal(msg.Value, []byte{0x00, 0xEE}) {
+		t.Error("Decoding produced incorrect message value.")
+	}
+}
diff --git a/functional_consumer_test.go b/functional_consumer_test.go
index 135ee79..8b31b45 100644
--- a/functional_consumer_test.go
+++ b/functional_consumer_test.go
@@ -106,6 +106,21 @@ func TestVersionMatrixLZ4(t *testing.T) {
 	consumeMsgs(t, testVersions, producedMessages)
 }
 
+// Support for zstd codec was introduced in v2.1.0.0
+func TestVersionMatrixZstd(t *testing.T) {
+	setupFunctionalTest(t)
+	defer teardownFunctionalTest(t)
+
+	// Produce lot's of message with all possible combinations of supported
+	// protocol versions starting with v2.1.0.0 (first where zstd was supported)
+	testVersions := versionRange(V2_1_0_0)
+	allCodecs := []CompressionCodec{CompressionZSTD}
+	producedMessages := produceMsgs(t, testVersions, allCodecs, 17, 100, false)
+
+	// When/Then
+	consumeMsgs(t, testVersions, producedMessages)
+}
+
 func TestVersionMatrixIdempotent(t *testing.T) {
 	setupFunctionalTest(t)
 	defer teardownFunctionalTest(t)
diff --git a/functional_producer_test.go b/functional_producer_test.go
index 5acd3b4..1fa0ba1 100644
--- a/functional_producer_test.go
+++ b/functional_producer_test.go
@@ -173,7 +173,6 @@ func testProducingMessages(t *testing.T, config *Config) {
 				t.Fatalf("Unexpected message with index %d: %s", i, message.Value)
 			}
 		}
-
 	}
 	safeClose(t, consumer)
 	safeClose(t, client)
@@ -261,6 +260,9 @@ func validateMetrics(t *testing.T, client Client) {
 		metricValidators.registerForBroker(broker, minValHistogramValidator("response-size", 1))
 	}
 
+	// There should be no requests in flight anymore
+	metricValidators.registerForAllBrokers(broker, counterValidator("requests-in-flight", 0))
+
 	// Run the validators
 	metricValidators.run(t, client.Config().MetricRegistry)
 }
diff --git a/functional_test.go b/functional_test.go
index 846eb29..778d9e0 100644
--- a/functional_test.go
+++ b/functional_test.go
@@ -23,10 +23,8 @@ var (
 	kafkaAvailable, kafkaRequired bool
 	kafkaBrokers                  []string
 
-	proxyClient  *toxiproxy.Client
-	Proxies      map[string]*toxiproxy.Proxy
-	ZKProxies    = []string{"zk1", "zk2", "zk3", "zk4", "zk5"}
-	KafkaProxies = []string{"kafka1", "kafka2", "kafka3", "kafka4", "kafka5"}
+	proxyClient *toxiproxy.Client
+	Proxies     map[string]*toxiproxy.Proxy
 )
 
 func init() {
diff --git a/heartbeat_request_test.go b/heartbeat_request_test.go
index da6cd18..981c377 100644
--- a/heartbeat_request_test.go
+++ b/heartbeat_request_test.go
@@ -11,9 +11,7 @@ var (
 )
 
 func TestHeartbeatRequest(t *testing.T) {
-	var request *HeartbeatRequest
-
-	request = new(HeartbeatRequest)
+	request := new(HeartbeatRequest)
 	request.GroupId = "foo"
 	request.GenerationId = 66051
 	request.MemberId = "baz"
diff --git a/heartbeat_response_test.go b/heartbeat_response_test.go
index 5bcbec9..4b29416 100644
--- a/heartbeat_response_test.go
+++ b/heartbeat_response_test.go
@@ -8,9 +8,7 @@ var (
 )
 
 func TestHeartbeatResponse(t *testing.T) {
-	var response *HeartbeatResponse
-
-	response = new(HeartbeatResponse)
+	response := new(HeartbeatResponse)
 	testVersionDecodable(t, "no error", response, heartbeatResponseNoError, 0)
 	if response.Err != ErrNoError {
 		t.Error("Decoding error failed: no error expected but found", response.Err)
diff --git a/kerberos_client_test.go b/kerberos_client_test.go
index 0072c0a..8af0e17 100644
--- a/kerberos_client_test.go
+++ b/kerberos_client_test.go
@@ -2,9 +2,10 @@ package sarama
 
 import (
 	"errors"
+	"testing"
+
 	krbcfg "gopkg.in/jcmturner/gokrb5.v7/config"
 	"gopkg.in/jcmturner/gokrb5.v7/test/testdata"
-	"testing"
 )
 
 /*
@@ -13,20 +14,6 @@ import (
  * properly.
  *
  */
-const TEST_KRB5CONF = `
-[libdefaults]
-default_realm = EXAMPLE.COM
-
-[realms]
-EXAMPLE.COM = {
-kdc = kerberos.example.com
-admin_server = kerberos.example.com
-}
-
-[domain_realm]
-.example.com = EXAMPLE.COM
-example.com = EXAMPLE.COM
-`
 
 func TestFaildToCreateKerberosConfig(t *testing.T) {
 	expectedErr := errors.New("configuration file could not be opened: krb5.conf open krb5.conf: no such file or directory")
diff --git a/leave_group_request_test.go b/leave_group_request_test.go
index c1fed6d..04527e4 100644
--- a/leave_group_request_test.go
+++ b/leave_group_request_test.go
@@ -10,9 +10,7 @@ var (
 )
 
 func TestLeaveGroupRequest(t *testing.T) {
-	var request *LeaveGroupRequest
-
-	request = new(LeaveGroupRequest)
+	request := new(LeaveGroupRequest)
 	request.GroupId = "foo"
 	request.MemberId = "bar"
 	testRequest(t, "basic", request, basicLeaveGroupRequest)
diff --git a/metadata_request_test.go b/metadata_request_test.go
index 0da69e0..16f6b59 100644
--- a/metadata_request_test.go
+++ b/metadata_request_test.go
@@ -104,6 +104,9 @@ func TestMetadataRequestV3(t *testing.T) {
 
 	request.Topics = []string{"topic1"}
 	testRequest(t, "one topic", request, metadataRequestOneTopicV3)
+
+	request.Topics = []string{"foo", "bar", "baz"}
+	testRequest(t, "three topics", request, metadataRequestThreeTopicsV3)
 }
 
 func TestMetadataRequestV4(t *testing.T) {
@@ -123,7 +126,7 @@ func TestMetadataRequestV4(t *testing.T) {
 func TestMetadataRequestV5(t *testing.T) {
 	request := new(MetadataRequest)
 	request.Version = 5
-	testRequest(t, "no topics", request, metadataRequestNoTopicsV4)
+	testRequest(t, "no topics", request, metadataRequestNoTopicsV5)
 
 	request.Topics = []string{"topic1"}
 
diff --git a/metrics_test.go b/metrics_test.go
index 789c0ff..7572f5b 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -170,3 +170,19 @@ func maxValHistogramValidator(name string, maxMax int) *metricValidator {
 		}
 	})
 }
+
+func counterValidator(name string, expectedCount int) *metricValidator {
+	return &metricValidator{
+		name: name,
+		validator: func(t *testing.T, metric interface{}) {
+			if counter, ok := metric.(metrics.Counter); !ok {
+				t.Errorf("Expected counter metric for '%s', got %T", name, metric)
+			} else {
+				count := counter.Count()
+				if count != int64(expectedCount) {
+					t.Errorf("Expected counter metric '%s' count = %d, got %d", name, expectedCount, count)
+				}
+			}
+		},
+	}
+}
diff --git a/offset_manager_test.go b/offset_manager_test.go
index 765db16..eca926d 100644
--- a/offset_manager_test.go
+++ b/offset_manager_test.go
@@ -9,7 +9,6 @@ import (
 func initOffsetManagerWithBackoffFunc(t *testing.T, retention time.Duration,
 	backoffFunc func(retries, maxRetries int) time.Duration, config *Config) (om OffsetManager,
 	testClient Client, broker, coordinator *MockBroker) {
-
 	config.Metadata.Retry.Max = 1
 	if backoffFunc != nil {
 		config.Metadata.Retry.BackoffFunc = backoffFunc
@@ -56,7 +55,6 @@ func initOffsetManager(t *testing.T, retention time.Duration) (om OffsetManager,
 
 func initPartitionOffsetManager(t *testing.T, om OffsetManager,
 	coordinator *MockBroker, initialOffset int64, metadata string) PartitionOffsetManager {
-
 	fetchResponse := new(OffsetFetchResponse)
 	fetchResponse.AddBlock("my_topic", 0, &OffsetFetchResponseBlock{
 		Err:      ErrNoError,
@@ -122,7 +120,6 @@ func TestNewOffsetManagerOffsetsAutoCommit(t *testing.T) {
 	// Tests to validate configuration of `Consumer.Offsets.AutoCommit.Enable`
 	for _, tt := range offsetsautocommitTestTable {
 		t.Run(tt.name, func(t *testing.T) {
-
 			config := NewConfig()
 			if tt.set {
 				config.Consumer.Offsets.AutoCommit.Enable = tt.enable
diff --git a/produce_response_test.go b/produce_response_test.go
index 197c7fb..0dfac79 100644
--- a/produce_response_test.go
+++ b/produce_response_test.go
@@ -10,8 +10,8 @@ var (
 	produceResponseNoBlocksV0 = []byte{
 		0x00, 0x00, 0x00, 0x00}
 
-	produceResponseManyBlocksVersions = [][]byte{
-		{
+	produceResponseManyBlocksVersions = map[int][]byte{
+		0: {
 			0x00, 0x00, 0x00, 0x01,
 
 			0x00, 0x03, 'f', 'o', 'o',
@@ -20,7 +20,9 @@ var (
 			0x00, 0x00, 0x00, 0x01, // Partition 1
 			0x00, 0x02, // ErrInvalidMessage
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
-		}, {
+		},
+
+		1: {
 			0x00, 0x00, 0x00, 0x01,
 
 			0x00, 0x03, 'f', 'o', 'o',
@@ -31,7 +33,8 @@ var (
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
 
 			0x00, 0x00, 0x00, 0x64, // 100 ms throttle time
-		}, {
+		},
+		2: {
 			0x00, 0x00, 0x00, 0x01,
 
 			0x00, 0x03, 'f', 'o', 'o',
@@ -42,6 +45,20 @@ var (
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0xE8, // Timestamp January 1st 0001 at 00:00:01,000 UTC (LogAppendTime was used)
 
+			0x00, 0x00, 0x00, 0x64, // 100 ms throttle time
+		},
+		7: { // version 7 adds StartOffset
+			0x00, 0x00, 0x00, 0x01,
+
+			0x00, 0x03, 'f', 'o', 'o',
+			0x00, 0x00, 0x00, 0x01,
+
+			0x00, 0x00, 0x00, 0x01, // Partition 1
+			0x00, 0x02, // ErrInvalidMessage
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0xE8, // Timestamp January 1st 0001 at 00:00:01,000 UTC (LogAppendTime was used)
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x32, // StartOffset 50
+
 			0x00, 0x00, 0x00, 0x64, // 100 ms throttle time
 		},
 	}
@@ -69,14 +86,19 @@ func TestProduceResponseDecode(t *testing.T) {
 			t.Error("Decoding did not produce a block for foo/1")
 		} else {
 			if block.Err != ErrInvalidMessage {
-				t.Error("Decoding failed for foo/2/Err, got:", int16(block.Err))
+				t.Error("Decoding failed for foo/1/Err, got:", int16(block.Err))
 			}
 			if block.Offset != 255 {
 				t.Error("Decoding failed for foo/1/Offset, got:", block.Offset)
 			}
 			if v >= 2 {
 				if block.Timestamp != time.Unix(1, 0) {
-					t.Error("Decoding failed for foo/2/Timestamp, got:", block.Timestamp)
+					t.Error("Decoding failed for foo/1/Timestamp, got:", block.Timestamp)
+				}
+			}
+			if v >= 7 {
+				if block.StartOffset != 50 {
+					t.Error("Decoding failed for foo/1/StartOffset, got:", block.StartOffset)
 				}
 			}
 		}
@@ -95,9 +117,10 @@ func TestProduceResponseEncode(t *testing.T) {
 
 	response.Blocks["foo"] = make(map[int32]*ProduceResponseBlock)
 	response.Blocks["foo"][1] = &ProduceResponseBlock{
-		Err:       ErrInvalidMessage,
-		Offset:    255,
-		Timestamp: time.Unix(1, 0),
+		Err:         ErrInvalidMessage,
+		Offset:      255,
+		Timestamp:   time.Unix(1, 0),
+		StartOffset: 50,
 	}
 	response.ThrottleTime = 100 * time.Millisecond
 	for v, produceResponseManyBlocks := range produceResponseManyBlocksVersions {
diff --git a/produce_set_test.go b/produce_set_test.go
index e517293..c092883 100644
--- a/produce_set_test.go
+++ b/produce_set_test.go
@@ -228,15 +228,15 @@ func TestProduceSetV3RequestBuilding(t *testing.T) {
 		Key:       StringEncoder(TestMessage),
 		Value:     StringEncoder(TestMessage),
 		Headers: []RecordHeader{
-			RecordHeader{
+			{
 				Key:   []byte("header-1"),
 				Value: []byte("value-1"),
 			},
-			RecordHeader{
+			{
 				Key:   []byte("header-2"),
 				Value: []byte("value-2"),
 			},
-			RecordHeader{
+			{
 				Key:   []byte("header-3"),
 				Value: []byte("value-3"),
 			},
@@ -306,15 +306,15 @@ func TestProduceSetIdempotentRequestBuilding(t *testing.T) {
 		Key:       StringEncoder(TestMessage),
 		Value:     StringEncoder(TestMessage),
 		Headers: []RecordHeader{
-			RecordHeader{
+			{
 				Key:   []byte("header-1"),
 				Value: []byte("value-1"),
 			},
-			RecordHeader{
+			{
 				Key:   []byte("header-2"),
 				Value: []byte("value-2"),
 			},
-			RecordHeader{
+			{
 				Key:   []byte("header-3"),
 				Value: []byte("value-3"),
 			},
diff --git a/request_test.go b/request_test.go
index ffc2bb7..fec1907 100644
--- a/request_test.go
+++ b/request_test.go
@@ -8,21 +8,6 @@ import (
 	"github.com/davecgh/go-spew/spew"
 )
 
-type testRequestBody struct {
-}
-
-func (s *testRequestBody) key() int16 {
-	return 0x666
-}
-
-func (s *testRequestBody) version() int16 {
-	return 0xD2
-}
-
-func (s *testRequestBody) encode(pe packetEncoder) error {
-	return pe.putString("abc")
-}
-
 // not specific to request tests, just helper functions for testing structures that
 // implement the encoder or decoder interfaces that needed somewhere to live
 
diff --git a/sasl_authenticate_request_test.go b/sasl_authenticate_request_test.go
index 0dee76b..fc059e0 100644
--- a/sasl_authenticate_request_test.go
+++ b/sasl_authenticate_request_test.go
@@ -9,9 +9,7 @@ var (
 )
 
 func TestSaslAuthenticateRequest(t *testing.T) {
-	var request *SaslAuthenticateRequest
-
-	request = new(SaslAuthenticateRequest)
+	request := new(SaslAuthenticateRequest)
 	request.SaslAuthBytes = []byte(`foo`)
 	testRequest(t, "basic", request, saslAuthenticateRequest)
 }
diff --git a/sasl_authenticate_response_test.go b/sasl_authenticate_response_test.go
index c232183..0444796 100644
--- a/sasl_authenticate_response_test.go
+++ b/sasl_authenticate_response_test.go
@@ -11,7 +11,6 @@ var (
 )
 
 func TestSaslAuthenticateResponse(t *testing.T) {
-
 	response := new(SaslAuthenticateResponse)
 	response.Err = ErrSASLAuthenticationFailed
 	msg := "err"
diff --git a/sasl_handshake_request_test.go b/sasl_handshake_request_test.go
index 806e628..34f954c 100644
--- a/sasl_handshake_request_test.go
+++ b/sasl_handshake_request_test.go
@@ -9,9 +9,7 @@ var (
 )
 
 func TestSaslHandshakeRequest(t *testing.T) {
-	var request *SaslHandshakeRequest
-
-	request = new(SaslHandshakeRequest)
+	request := new(SaslHandshakeRequest)
 	request.Mechanism = "foo"
 	testRequest(t, "basic", request, baseSaslRequest)
 }
diff --git a/sasl_handshake_response_test.go b/sasl_handshake_response_test.go
index 1fd4c79..0ebc27a 100644
--- a/sasl_handshake_response_test.go
+++ b/sasl_handshake_response_test.go
@@ -11,9 +11,7 @@ var (
 )
 
 func TestSaslHandshakeResponse(t *testing.T) {
-	var response *SaslHandshakeResponse
-
-	response = new(SaslHandshakeResponse)
+	response := new(SaslHandshakeResponse)
 	testVersionDecodable(t, "no error", response, saslHandshakeResponse, 0)
 	if response.Err != ErrNoError {
 		t.Error("Decoding error failed: no error expected but found", response.Err)
diff --git a/txn_offset_commit_request_test.go b/txn_offset_commit_request_test.go
index fe3bba5..f60d417 100644
--- a/txn_offset_commit_request_test.go
+++ b/txn_offset_commit_request_test.go
@@ -24,7 +24,7 @@ func TestTxnOffsetCommitRequest(t *testing.T) {
 		ProducerID:      8000,
 		ProducerEpoch:   1,
 		Topics: map[string][]*PartitionOffsetMetadata{
-			"topic": []*PartitionOffsetMetadata{{
+			"topic": {{
 				Offset:    123,
 				Partition: 2,
 			}},
diff --git a/txn_offset_commit_response_test.go b/txn_offset_commit_response_test.go
index b275265..2a66ad3 100644
--- a/txn_offset_commit_response_test.go
+++ b/txn_offset_commit_response_test.go
@@ -20,7 +20,7 @@ func TestTxnOffsetCommitResponse(t *testing.T) {
 	resp := &TxnOffsetCommitResponse{
 		ThrottleTime: 100 * time.Millisecond,
 		Topics: map[string][]*PartitionError{
-			"topic": []*PartitionError{{
+			"topic": {{
 				Partition: 2,
 				Err:       ErrInvalidProducerEpoch,
 			}},