Repository: Shopify/sarama
Pull request #1594 (2020-01-31 15:06:50)

Sets ConfigEntry.Default flag in addition to the ConfigEntry.Source for Kafka versions > V1_1_0_0
We were relying on the `ConfigEntry.Default` field to filter out any default config settings when using `admin.DescribeConfig()`.

After updating to the most recent version of sarama, this flag is always set to `false` with Kafka versions > `V1_1_0_0`.

We found out that we can now use the `ConfigEntry.Source` to check for defaults. I don't know it was intended change to "break" the `ConfigEntry.Default` in higher versions - let me know. 

This change sets the `ConfigEntry.Default` flag in addition to the `ConfigEntry.Source` for Kafka versions > `V1_1_0_0`. 

I haven't written any tests yet as I'm not familiar with the code base. I can add some later.


Patch:
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
index bffdb9d..a4a14c6 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -1,6 +1,6 @@
 name: CI
 
-on: push
+on: [push, pull_request]
 
 jobs:
   test:
@@ -46,7 +46,7 @@ jobs:
 
     - 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.1
+        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
diff --git a/.golangci.yml b/.golangci.yml
index 561c2c2..47624f3 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -38,32 +38,32 @@ linters:
   disable-all: true
   enable:
     - bodyclose
-    # - deadcode
+    - deadcode
     - depguard
     - dogsled
     # - dupl
-    # - errcheck
+    - errcheck
     - funlen
     # - gocritic
     - gocyclo
     - gofmt
-    # - goimports
+    - goimports
     # - golint
-    # - gosec
+    - gosec
     # - gosimple
     - govet
     # - ineffassign
-    # - interfacer
+    - interfacer
     # - misspell
     # - nakedret
     # - scopelint
     # - staticcheck
-    # - structcheck
+    - structcheck
     # - stylecheck
     - typecheck
-    # - unconvert
-    # - unused
-    # - varcheck
+    - unconvert
+    - unused
+    - varcheck
     - whitespace
     # - goconst
     # - gochecknoinits
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f728a67..2bebeb1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,8 +1,19 @@
 # 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:
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 dd63484..7dd7250 100644
--- a/admin.go
+++ b/admin.go
@@ -338,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
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 de83987..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 {
@@ -712,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
 	}
 
@@ -816,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
 		}
@@ -891,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
 	}
@@ -902,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
 	}
@@ -910,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
 	}
@@ -981,9 +995,12 @@ func (b *Broker) sendAndReceiveV0SASLPlainAuth() error {
 	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
 	}
@@ -1008,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
 	}
@@ -1066,14 +1085,17 @@ func (b *Broker) sendAndReceiveSASLOAuth(provider AccessTokenProvider) error {
 // 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,17 +1130,21 @@ func (b *Broker) sendAndReceiveSASLSCRAMv1() 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
 		}
@@ -1271,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 {
@@ -1290,13 +1316,22 @@ 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)
 
 	if b.brokerRequestLatency != nil {
 		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) {
@@ -1325,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 {
@@ -1344,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 057a57a..81a3ca5 100644
--- a/client.go
+++ b/client.go
@@ -798,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..8102578 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"
 )
 
diff --git a/config.go b/config.go
index 7495200..08d50ce 100644
--- a/config.go
+++ b/config.go
@@ -396,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.
diff --git a/consumer.go b/consumer.go
index ff44ade..e311c7e 100644
--- a/consumer.go
+++ b/consumer.go
@@ -887,10 +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 fc95cd0..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.
@@ -331,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
 		}
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 a18eeba..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()
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 836e6de..9a3e8dd 100644
--- a/fetch_request.go
+++ b/fetch_request.go
@@ -55,6 +55,7 @@ type FetchRequest struct {
 	SessionEpoch int32
 	blocks       map[string]map[int32]*fetchRequestBlock
 	forgotten    map[string][]int32
+	RackID       string
 }
 
 type IsolationLevel int8
@@ -118,6 +119,12 @@ func (r *FetchRequest) encode(pe packetEncoder) (err error) {
 			}
 		}
 	}
+	if r.Version >= 11 {
+		err = pe.putString(r.RackID)
+		if err != nil {
+			return err
+		}
+	}
 
 	return nil
 }
@@ -192,9 +199,6 @@ func (r *FetchRequest) decode(pd packetDecoder, version int16) (err error) {
 		if err != nil {
 			return err
 		}
-		if forgottenCount == 0 {
-			return nil
-		}
 		r.forgotten = make(map[string][]int32)
 		for i := 0; i < forgottenCount; i++ {
 			topic, err := pd.getString()
@@ -217,6 +221,13 @@ func (r *FetchRequest) decode(pd packetDecoder, version int16) (err error) {
 		}
 	}
 
+	if r.Version >= 11 {
+		r.RackID, err = pd.getString()
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
diff --git a/fetch_response.go b/fetch_response.go
index 26936d9..dc0aeed 100644
--- a/fetch_response.go
+++ b/fetch_response.go
@@ -30,14 +30,15 @@ func (t *AbortedTransaction) encode(pe packetEncoder) (err error) {
 }
 
 type FetchResponseBlock struct {
-	Err                 KError
-	HighWaterMarkOffset int64
-	LastStableOffset    int64
-	LogStartOffset      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) {
@@ -83,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
@@ -188,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)
diff --git a/go.mod b/go.mod
index 8ba2c91..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.9.7
-	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 7f61258..06ec328 100644
--- a/go.sum
+++ b/go.sum
@@ -1,69 +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/klauspost/compress v1.9.7 h1:hYW1gP94JUmAhBtJ+LNz5My+gBobDxPR1iVuKug26aA=
-github.com/klauspost/compress v1.9.7/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 32ca93d..98be0f5 100644
--- a/gssapi_kerberos.go
+++ b/gssapi_kerberos.go
@@ -242,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/mockbroker.go b/mockbroker.go
index cd1a850..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()
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 984e5aa..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
 }
 
@@ -736,6 +736,7 @@ func (mr *MockDescribeConfigsResponse) For(reqBody versionedDecoder) encoder {
 	}
 
 	includeSynonyms := (req.Version > 0)
+	includeSource := (req.Version > 0)
 
 	for _, r := range req.Resources {
 		var configEntries []*ConfigEntry
@@ -770,9 +771,12 @@ func (mr *MockDescribeConfigsResponse) For(reqBody versionedDecoder) encoder {
 			maxMessageBytes := &ConfigEntry{Name: "max.message.bytes",
 				Value:     "1000000",
 				ReadOnly:  false,
-				Default:   true,
+				Default:   !includeSource,
 				Sensitive: false,
 			}
+			if includeSource {
+				maxMessageBytes.Source = SourceDefault
+			}
 			if includeSynonyms {
 				maxMessageBytes.Synonyms = []*ConfigSynonym{
 					{
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_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_response.go b/produce_response.go
index e4f19a7..7fc2fff 100644
--- a/produce_response.go
+++ b/produce_response.go
@@ -163,14 +163,6 @@ func (r *ProduceResponse) encode(pe packetEncoder) error {
 	return nil
 }
 
-func (r *ProduceResponse) key() int16 {
-	return 0
-}
-
-func (r *ProduceResponse) version() int16 {
-	return r.Version
-}
-
 func (r *ProduceResponse) GetBlock(topic string, partition int32) *ProduceResponseBlock {
 	if r.Blocks == nil {
 		return nil
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/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 19a9e98..b88e1f1 100644
--- a/tools/kafka-console-producer/kafka-console-producer.go
+++ b/tools/kafka-console-producer/kafka-console-producer.go
@@ -8,9 +8,10 @@ import (
 	"os"
 	"strings"
 
+	"github.com/rcrowley/go-metrics"
+
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
-	"github.com/rcrowley/go-metrics"
 )
 
 var (
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/admin_test.go b/admin_test.go
index 113f69b..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"))
 	}
@@ -615,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)
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 7aeff10..3de3086 100644
--- a/async_producer_test.go
+++ b/async_producer_test.go
@@ -311,22 +311,6 @@ 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)
diff --git a/balance_strategy_test.go b/balance_strategy_test.go
index e9c7274..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
@@ -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
@@ -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()))
diff --git a/broker_test.go b/broker_test.go
index 224971a..387dae9 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -217,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
@@ -244,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 {
@@ -335,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})
@@ -453,6 +451,7 @@ 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
@@ -536,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()
@@ -626,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"
@@ -990,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/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/describe_configs_request_test.go b/describe_configs_request_test.go
index ee78f14..ef9d3ff 100644
--- a/describe_configs_request_test.go
+++ b/describe_configs_request_test.go
@@ -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 90fcf25..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) {
@@ -86,6 +117,7 @@ func TestDescribeConfigsResponsev0(t *testing.T) {
 						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
 
@@ -119,6 +185,7 @@ func TestDescribeConfigsResponsev1(t *testing.T) {
 						Value:     "1000",
 						ReadOnly:  false,
 						Source:    SourceStaticBroker,
+						Default:   false,
 						Sensitive: false,
 						Synonyms:  []*ConfigSynonym{},
 					},
@@ -154,6 +221,7 @@ func TestDescribeConfigsResponseWithSynonym(t *testing.T) {
 						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/fetch_request_test.go b/fetch_request_test.go
index 2fdd905..0903caf 100644
--- a/fetch_request_test.go
+++ b/fetch_request_test.go
@@ -26,6 +26,24 @@ 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) {
@@ -57,4 +75,16 @@ func TestFetchRequest(t *testing.T) {
 		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 3801257..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) {
@@ -398,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_producer_test.go b/functional_producer_test.go
index 311de0a..1fa0ba1 100644
--- a/functional_producer_test.go
+++ b/functional_producer_test.go
@@ -260,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/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_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)