Repository: Shopify/sarama
Pull request #1666 (2020-04-15 07:14:54)

Bugfix: Allow TLS connections to work over socks proxy.
It seems that the code was either proxy or tls but in my case I need tls over the socks proxy so Ive created this PR to fix this issue.
Note there seems to be an issue with the tls test where the server name was never set so Ive fixed that too.


Patch:
diff --git a/acl_create_response.go b/acl_create_response.go
index bc018ed..14b1b9e 100644
--- a/acl_create_response.go
+++ b/acl_create_response.go
@@ -2,7 +2,7 @@ package sarama
 
 import "time"
 
-//CreateAclsResponse is a an acl reponse creation type
+//CreateAclsResponse is a an acl response creation type
 type CreateAclsResponse struct {
 	ThrottleTime         time.Duration
 	AclCreationResponses []*AclCreationResponse
diff --git a/alter_configs_response.go b/alter_configs_response.go
index 99a5260..3266f92 100644
--- a/alter_configs_response.go
+++ b/alter_configs_response.go
@@ -2,13 +2,13 @@ package sarama
 
 import "time"
 
-//AlterConfigsResponse is a reponse type for alter config
+//AlterConfigsResponse is a response type for alter config
 type AlterConfigsResponse struct {
 	ThrottleTime time.Duration
 	Resources    []*AlterConfigsResourceResponse
 }
 
-//AlterConfigsResourceResponse is a reponse type for alter config resource
+//AlterConfigsResourceResponse is a response type for alter config resource
 type AlterConfigsResourceResponse struct {
 	ErrorCode int16
 	ErrorMsg  string
diff --git a/api_versions_response.go b/api_versions_response.go
index 582e29d..d09e8d9 100644
--- a/api_versions_response.go
+++ b/api_versions_response.go
@@ -1,6 +1,6 @@
 package sarama
 
-//ApiVersionsResponseBlock is an api version reponse block type
+//ApiVersionsResponseBlock is an api version response block type
 type ApiVersionsResponseBlock struct {
 	ApiKey     int16
 	MinVersion int16
diff --git a/async_producer.go b/async_producer.go
index 5c57a81..d0ce01b 100644
--- a/async_producer.go
+++ b/async_producer.go
@@ -60,13 +60,28 @@ const (
 	noProducerEpoch = -1
 )
 
-func (t *transactionManager) getAndIncrementSequenceNumber(topic string, partition int32) int32 {
+func (t *transactionManager) getAndIncrementSequenceNumber(topic string, partition int32) (int32, int16) {
 	key := fmt.Sprintf("%s-%d", topic, partition)
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 	sequence := t.sequenceNumbers[key]
 	t.sequenceNumbers[key] = sequence + 1
-	return sequence
+	return sequence, t.producerEpoch
+}
+
+func (t *transactionManager) bumpEpoch() {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	t.producerEpoch++
+	for k := range t.sequenceNumbers {
+		t.sequenceNumbers[k] = 0
+	}
+}
+
+func (t *transactionManager) getProducerID() (int64, int16) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	return t.producerID, t.producerEpoch
 }
 
 func newTransactionManager(conf *Config, client Client) (*transactionManager, error) {
@@ -208,6 +223,8 @@ type ProducerMessage struct {
 	flags          flagSet
 	expectation    chan *ProducerError
 	sequenceNumber int32
+	producerEpoch  int16
+	hasSequence    bool
 }
 
 const producerMessageOverhead = 26 // the metadata overhead of CRC, flags, etc.
@@ -234,6 +251,9 @@ func (m *ProducerMessage) byteSize(version int) int {
 func (m *ProducerMessage) clear() {
 	m.flags = 0
 	m.retries = 0
+	m.sequenceNumber = 0
+	m.producerEpoch = 0
+	m.hasSequence = false
 }
 
 // ProducerError is the type of error generated when the producer fails to deliver a message.
@@ -388,10 +408,6 @@ func (tp *topicProducer) dispatch() {
 				continue
 			}
 		}
-		// All messages being retried (sent or not) have already had their retry count updated
-		if tp.parent.conf.Producer.Idempotent && msg.retries == 0 {
-			msg.sequenceNumber = tp.parent.txnmgr.getAndIncrementSequenceNumber(msg.Topic, msg.Partition)
-		}
 
 		handler := tp.handlers[msg.Partition]
 		if handler == nil {
@@ -570,6 +586,15 @@ func (pp *partitionProducer) dispatch() {
 			Logger.Printf("producer/leader/%s/%d selected broker %d\n", pp.topic, pp.partition, pp.leader.ID())
 		}
 
+		// Now that we know we have a broker to actually try and send this message to, generate the sequence
+		// number for it.
+		// All messages being retried (sent or not) have already had their retry count updated
+		// Also, ignore "special" syn/fin messages used to sync the brokerProducer and the topicProducer.
+		if pp.parent.conf.Producer.Idempotent && msg.retries == 0 && msg.flags == 0 {
+			msg.sequenceNumber, msg.producerEpoch = pp.parent.txnmgr.getAndIncrementSequenceNumber(msg.Topic, msg.Partition)
+			msg.hasSequence = true
+		}
+
 		pp.brokerProducer.input <- msg
 	}
 }
@@ -748,12 +773,21 @@ func (bp *brokerProducer) run() {
 			}
 
 			if bp.buffer.wouldOverflow(msg) {
-				if err := bp.waitForSpace(msg); err != nil {
+				Logger.Printf("producer/broker/%d maximum request accumulated, waiting for space\n", bp.broker.ID())
+				if err := bp.waitForSpace(msg, false); err != nil {
 					bp.parent.retryMessage(msg, err)
 					continue
 				}
 			}
 
+			if bp.parent.txnmgr.producerID != noProducerID && bp.buffer.producerEpoch != msg.producerEpoch {
+				// The epoch was reset, need to roll the buffer over
+				Logger.Printf("producer/broker/%d detected epoch rollover, waiting for new buffer\n", bp.broker.ID())
+				if err := bp.waitForSpace(msg, true); err != nil {
+					bp.parent.retryMessage(msg, err)
+					continue
+				}
+			}
 			if err := bp.buffer.add(msg); err != nil {
 				bp.parent.returnError(msg, err)
 				continue
@@ -809,9 +843,7 @@ func (bp *brokerProducer) needsRetry(msg *ProducerMessage) error {
 	return bp.currentRetries[msg.Topic][msg.Partition]
 }
 
-func (bp *brokerProducer) waitForSpace(msg *ProducerMessage) error {
-	Logger.Printf("producer/broker/%d maximum request accumulated, waiting for space\n", bp.broker.ID())
-
+func (bp *brokerProducer) waitForSpace(msg *ProducerMessage, forceRollover bool) error {
 	for {
 		select {
 		case response := <-bp.responses:
@@ -819,7 +851,7 @@ func (bp *brokerProducer) waitForSpace(msg *ProducerMessage) error {
 			// handling a response can change our state, so re-check some things
 			if reason := bp.needsRetry(msg); reason != nil {
 				return reason
-			} else if !bp.buffer.wouldOverflow(msg) {
+			} else if !bp.buffer.wouldOverflow(msg) && !forceRollover {
 				return nil
 			}
 		case bp.output <- bp.buffer:
@@ -1030,6 +1062,12 @@ func (p *asyncProducer) shutdown() {
 }
 
 func (p *asyncProducer) returnError(msg *ProducerMessage, err error) {
+	// We need to reset the producer ID epoch if we set a sequence number on it, because the broker
+	// will never see a message with this number, so we can never continue the sequence.
+	if msg.hasSequence {
+		Logger.Printf("producer/txnmanager rolling over epoch due to publish failure on %s/%d", msg.Topic, msg.Partition)
+		p.txnmgr.bumpEpoch()
+	}
 	msg.clear()
 	pErr := &ProducerError{Msg: msg, Err: err}
 	if p.conf.Producer.Return.Errors {
diff --git a/broker.go b/broker.go
index 4f3991a..f379dca 100644
--- a/broker.go
+++ b/broker.go
@@ -73,7 +73,7 @@ const (
 	// server negotiate SASL by wrapping tokens with Kafka protocol headers.
 	SASLHandshakeV1 = int16(1)
 	// SASLExtKeyAuth is the reserved extension key name sent as part of the
-	// SASL/OAUTHBEARER intial client response
+	// SASL/OAUTHBEARER initial client response
 	SASLExtKeyAuth = "auth"
 )
 
@@ -154,25 +154,20 @@ func (b *Broker) Open(conf *Config) error {
 	go withRecover(func() {
 		defer b.lock.Unlock()
 
-		dialer := net.Dialer{
-			Timeout:   conf.Net.DialTimeout,
-			KeepAlive: conf.Net.KeepAlive,
-			LocalAddr: conf.Net.LocalAddr,
-		}
-
-		if conf.Net.TLS.Enable {
-			b.conn, b.connErr = tls.DialWithDialer(&dialer, "tcp", b.addr, conf.Net.TLS.Config)
-		} else if conf.Net.Proxy.Enable {
-			b.conn, b.connErr = conf.Net.Proxy.Dialer.Dial("tcp", b.addr)
-		} else {
-			b.conn, b.connErr = dialer.Dial("tcp", b.addr)
-		}
+		dialer := conf.getDialer()
+		b.conn, b.connErr = dialer.Dial("tcp", b.addr)
 		if b.connErr != nil {
 			Logger.Printf("Failed to connect to broker %s: %s\n", b.addr, b.connErr)
 			b.conn = nil
 			atomic.StoreInt32(&b.opened, 0)
 			return
 		}
+
+		if conf.Net.TLS.Enable {
+			Logger.Printf("Using tls")
+			b.conn = tls.Client(b.conn, conf.Net.TLS.Config)
+		}
+
 		b.conn = newBufConn(b.conn)
 
 		b.conf = conf
@@ -369,7 +364,7 @@ func (b *Broker) Fetch(request *FetchRequest) (*FetchResponse, error) {
 	return response, nil
 }
 
-//CommitOffset return an Offset commit reponse or error
+//CommitOffset return an Offset commit response or error
 func (b *Broker) CommitOffset(request *OffsetCommitRequest) (*OffsetCommitResponse, error) {
 	response := new(OffsetCommitResponse)
 
@@ -1014,7 +1009,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 // When credentials are invalid, Kafka replies with a SaslAuthenticate response
 // containing an error code and message detailing the authentication failure.
 func (b *Broker) sendAndReceiveSASLPlainAuth() error {
-	// default to V0 to allow for backward compatability when SASL is enabled
+	// default to V0 to allow for backward compatibility when SASL is enabled
 	// but not the handshake
 	if b.conf.Net.SASL.Handshake {
 		handshakeErr := b.sendAndReceiveSASLHandshake(SASLTypePlaintext, b.conf.Net.SASL.Version)
diff --git a/config.go b/config.go
index e899820..0ce308f 100644
--- a/config.go
+++ b/config.go
@@ -734,3 +734,16 @@ func (c *Config) Validate() error {
 
 	return nil
 }
+
+func (c *Config) getDialer() proxy.Dialer {
+	if c.Net.Proxy.Enable {
+		Logger.Printf("using proxy %s", c.Net.Proxy.Dialer)
+		return c.Net.Proxy.Dialer
+	} else {
+		return &net.Dialer{
+			Timeout:   c.Net.DialTimeout,
+			KeepAlive: c.Net.KeepAlive,
+			LocalAddr: c.Net.LocalAddr,
+		}
+	}
+}
diff --git a/examples/consumergroup/main.go b/examples/consumergroup/main.go
index 776448f..9a8b7cd 100644
--- a/examples/consumergroup/main.go
+++ b/examples/consumergroup/main.go
@@ -28,7 +28,7 @@ func init() {
 	flag.StringVar(&brokers, "brokers", "", "Kafka bootstrap brokers to connect to, as a comma separated list")
 	flag.StringVar(&group, "group", "", "Kafka consumer group definition")
 	flag.StringVar(&version, "version", "2.1.1", "Kafka cluster version")
-	flag.StringVar(&topics, "topics", "", "Kafka topics to be consumed, as a comma seperated list")
+	flag.StringVar(&topics, "topics", "", "Kafka topics to be consumed, as a comma separated list")
 	flag.StringVar(&assignor, "assignor", "range", "Consumer group partition assignment strategy (range, roundrobin, sticky)")
 	flag.BoolVar(&oldest, "oldest", true, "Kafka consumer consume initial offset from oldest")
 	flag.BoolVar(&verbose, "verbose", false, "Sarama logging")
diff --git a/produce_set.go b/produce_set.go
index 36c43c6..9c70f81 100644
--- a/produce_set.go
+++ b/produce_set.go
@@ -13,17 +13,22 @@ type partitionSet struct {
 }
 
 type produceSet struct {
-	parent *asyncProducer
-	msgs   map[string]map[int32]*partitionSet
+	parent        *asyncProducer
+	msgs          map[string]map[int32]*partitionSet
+	producerID    int64
+	producerEpoch int16
 
 	bufferBytes int
 	bufferCount int
 }
 
 func newProduceSet(parent *asyncProducer) *produceSet {
+	pid, epoch := parent.txnmgr.getProducerID()
 	return &produceSet{
-		msgs:   make(map[string]map[int32]*partitionSet),
-		parent: parent,
+		msgs:          make(map[string]map[int32]*partitionSet),
+		parent:        parent,
+		producerID:    pid,
+		producerEpoch: epoch,
 	}
 }
 
@@ -65,8 +70,8 @@ func (ps *produceSet) add(msg *ProducerMessage) error {
 				Version:          2,
 				Codec:            ps.parent.conf.Producer.Compression,
 				CompressionLevel: ps.parent.conf.Producer.CompressionLevel,
-				ProducerID:       ps.parent.txnmgr.producerID,
-				ProducerEpoch:    ps.parent.txnmgr.producerEpoch,
+				ProducerID:       ps.producerID,
+				ProducerEpoch:    ps.producerEpoch,
 			}
 			if ps.parent.conf.Producer.Idempotent {
 				batch.FirstSequence = msg.sequenceNumber
@@ -78,12 +83,17 @@ func (ps *produceSet) add(msg *ProducerMessage) error {
 		}
 		partitions[msg.Partition] = set
 	}
-	set.msgs = append(set.msgs, msg)
 
 	if ps.parent.conf.Version.IsAtLeast(V0_11_0_0) {
 		if ps.parent.conf.Producer.Idempotent && msg.sequenceNumber < set.recordsToSend.RecordBatch.FirstSequence {
 			return errors.New("assertion failed: message out of sequence added to a batch")
 		}
+	}
+
+	// Past this point we can't return an error, because we've already added the message to the set.
+	set.msgs = append(set.msgs, msg)
+
+	if ps.parent.conf.Version.IsAtLeast(V0_11_0_0) {
 		// We are being conservative here to avoid having to prep encode the record
 		size += maximumRecordOverhead
 		rec := &Record{
diff --git a/tools/kafka-console-producer/README.md b/tools/kafka-console-producer/README.md
index 6b3a65f..7802b8b 100644
--- a/tools/kafka-console-producer/README.md
+++ b/tools/kafka-console-producer/README.md
@@ -25,7 +25,7 @@ A simple command line tool to produce a single message to Kafka.
     # Partitioning: by default, kafka-console-producer will partition as follows:
     # - manual partitioning if a -partition is provided
     # - hash partitioning by key if a -key is provided
-    # - random partioning otherwise.
+    # - random partitioning otherwise.
     #
     # You can override this using the -partitioner argument:
     echo "hello world" | kafka-console-producer -topic=test -key=key -partitioner=random


Test patch:
diff --git a/alter_partition_reassignments_request_test.go b/alter_partition_reassignments_request_test.go
index 8d28272..c917f2d 100644
--- a/alter_partition_reassignments_request_test.go
+++ b/alter_partition_reassignments_request_test.go
@@ -4,13 +4,13 @@ import "testing"
 
 var (
 	alterPartitionReassignmentsRequestNoBlock = []byte{
-		0, 0, 39, 16, // timout 10000
+		0, 0, 39, 16, // timeout 10000
 		1, // 1-1=0 blocks
 		0, // empty tagged fields
 	}
 
 	alterPartitionReassignmentsRequestOneBlock = []byte{
-		0, 0, 39, 16, // timout 10000
+		0, 0, 39, 16, // timeout 10000
 		2,                         // 2-1=1 block
 		6, 116, 111, 112, 105, 99, // topic name "topic" as compact string
 		2,          // 2-1=1 partitions
@@ -22,7 +22,7 @@ var (
 	}
 
 	alterPartitionReassignmentsAbortRequest = []byte{
-		0, 0, 39, 16, // timout 10000
+		0, 0, 39, 16, // timeout 10000
 		2,                         // 2-1=1 block
 		6, 116, 111, 112, 105, 99, // topic name "topic" as compact string
 		2,          // 2-1=1 partitions
diff --git a/async_producer_test.go b/async_producer_test.go
index d0f012d..46b9779 100644
--- a/async_producer_test.go
+++ b/async_producer_test.go
@@ -1130,6 +1130,75 @@ func TestAsyncProducerIdempotentErrorOnOutOfSeq(t *testing.T) {
 	closeProducer(t, producer)
 }
 
+func TestAsyncProducerIdempotentEpochRollover(t *testing.T) {
+	broker := NewMockBroker(t, 1)
+	defer broker.Close()
+
+	metadataResponse := &MetadataResponse{
+		Version:      1,
+		ControllerID: 1,
+	}
+	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
+	metadataResponse.AddTopicPartition("my_topic", 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
+	broker.Returns(metadataResponse)
+
+	initProducerID := &InitProducerIDResponse{
+		ThrottleTime:  0,
+		ProducerID:    1000,
+		ProducerEpoch: 1,
+	}
+	broker.Returns(initProducerID)
+
+	config := NewConfig()
+	config.Producer.Flush.Messages = 10
+	config.Producer.Flush.Frequency = 10 * time.Millisecond
+	config.Producer.Return.Successes = true
+	config.Producer.Retry.Max = 1 // This test needs to exercise what happens when retries exhaust
+	config.Producer.RequiredAcks = WaitForAll
+	config.Producer.Retry.Backoff = 0
+	config.Producer.Idempotent = true
+	config.Net.MaxOpenRequests = 1
+	config.Version = V0_11_0_0
+
+	producer, err := NewAsyncProducer([]string{broker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer closeProducer(t, producer)
+
+	producer.Input() <- &ProducerMessage{Topic: "my_topic", Value: StringEncoder("hello")}
+	prodError := &ProduceResponse{
+		Version:      3,
+		ThrottleTime: 0,
+	}
+	prodError.AddTopicPartition("my_topic", 0, ErrBrokerNotAvailable)
+	broker.Returns(prodError)
+	<-producer.Errors()
+
+	lastReqRes := broker.history[len(broker.history)-1]
+	lastProduceBatch := lastReqRes.Request.(*ProduceRequest).records["my_topic"][0].RecordBatch
+	if lastProduceBatch.FirstSequence != 0 {
+		t.Error("first sequence not zero")
+	}
+	if lastProduceBatch.ProducerEpoch != 1 {
+		t.Error("first epoch was not one")
+	}
+
+	// Now if we produce again, the epoch should have rolled over.
+	producer.Input() <- &ProducerMessage{Topic: "my_topic", Value: StringEncoder("hello")}
+	broker.Returns(prodError)
+	<-producer.Errors()
+
+	lastReqRes = broker.history[len(broker.history)-1]
+	lastProduceBatch = lastReqRes.Request.(*ProduceRequest).records["my_topic"][0].RecordBatch
+	if lastProduceBatch.FirstSequence != 0 {
+		t.Error("second sequence not zero")
+	}
+	if lastProduceBatch.ProducerEpoch <= 1 {
+		t.Error("second epoch was not > 1")
+	}
+}
+
 // TestBrokerProducerShutdown ensures that a call to shutdown stops the
 // brokerProducer run() loop and doesn't leak any goroutines
 func TestBrokerProducerShutdown(t *testing.T) {
diff --git a/broker_test.go b/broker_test.go
index e2b1746..c7d5486 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -80,38 +80,40 @@ func TestBrokerAccessors(t *testing.T) {
 
 func TestSimpleBrokerCommunication(t *testing.T) {
 	for _, tt := range brokerTestTable {
-		Logger.Printf("Testing broker communication for %s", tt.name)
-		mb := NewMockBroker(t, 0)
-		mb.Returns(&mockEncoder{tt.response})
-		pendingNotify := make(chan brokerMetrics)
-		// Register a callback to be notified about successful requests
-		mb.SetNotifier(func(bytesRead, bytesWritten int) {
-			pendingNotify <- brokerMetrics{bytesRead, bytesWritten}
+		t.Run(tt.name, func(t *testing.T) {
+			Logger.Printf("Testing broker communication for %s", tt.name)
+			mb := NewMockBroker(t, 0)
+			mb.Returns(&mockEncoder{tt.response})
+			pendingNotify := make(chan brokerMetrics)
+			// Register a callback to be notified about successful requests
+			mb.SetNotifier(func(bytesRead, bytesWritten int) {
+				pendingNotify <- brokerMetrics{bytesRead, bytesWritten}
+			})
+			broker := NewBroker(mb.Addr())
+			// Set the broker id in order to validate local broker metrics
+			broker.id = 0
+			conf := NewConfig()
+			conf.Version = tt.version
+			err := broker.Open(conf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			tt.runner(t, broker)
+			// Wait up to 500 ms for the remote broker to process the request and
+			// notify us about the metrics
+			timeout := 500 * time.Millisecond
+			select {
+			case mockBrokerMetrics := <-pendingNotify:
+				validateBrokerMetrics(t, broker, mockBrokerMetrics)
+			case <-time.After(timeout):
+				t.Errorf("No request received for: %s after waiting for %v", tt.name, timeout)
+			}
+			mb.Close()
+			err = broker.Close()
+			if err != nil {
+				t.Error(err)
+			}
 		})
-		broker := NewBroker(mb.Addr())
-		// Set the broker id in order to validate local broker metrics
-		broker.id = 0
-		conf := NewConfig()
-		conf.Version = tt.version
-		err := broker.Open(conf)
-		if err != nil {
-			t.Fatal(err)
-		}
-		tt.runner(t, broker)
-		// Wait up to 500 ms for the remote broker to process the request and
-		// notify us about the metrics
-		timeout := 500 * time.Millisecond
-		select {
-		case mockBrokerMetrics := <-pendingNotify:
-			validateBrokerMetrics(t, broker, mockBrokerMetrics)
-		case <-time.After(timeout):
-			t.Errorf("No request received for: %s after waiting for %v", tt.name, timeout)
-		}
-		mb.Close()
-		err = broker.Close()
-		if err != nil {
-			t.Error(err)
-		}
 	}
 }
 
@@ -204,58 +206,60 @@ func TestSASLOAuthBearer(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-		// mockBroker mocks underlying network logic and broker responses
-		mockBroker := NewMockBroker(t, 0)
+		t.Run(test.name, func(t *testing.T) {
+			// mockBroker mocks underlying network logic and broker responses
+			mockBroker := NewMockBroker(t, 0)
 
-		mockBroker.SetHandlerByMap(map[string]MockResponse{
-			"SaslAuthenticateRequest": test.mockSASLAuthResponse,
-			"SaslHandshakeRequest":    test.mockSASLHandshakeResponse,
-		})
+			mockBroker.SetHandlerByMap(map[string]MockResponse{
+				"SaslAuthenticateRequest": test.mockSASLAuthResponse,
+				"SaslHandshakeRequest":    test.mockSASLHandshakeResponse,
+			})
 
-		// broker executes SASL requests against mockBroker
-		broker := NewBroker(mockBroker.Addr())
-		broker.requestRate = metrics.NilMeter{}
-		broker.outgoingByteRate = metrics.NilMeter{}
-		broker.incomingByteRate = metrics.NilMeter{}
-		broker.requestSize = metrics.NilHistogram{}
-		broker.responseSize = metrics.NilHistogram{}
-		broker.responseRate = metrics.NilMeter{}
-		broker.requestLatency = metrics.NilHistogram{}
-		broker.requestsInFlight = metrics.NilCounter{}
+			// broker executes SASL requests against mockBroker
+			broker := NewBroker(mockBroker.Addr())
+			broker.requestRate = metrics.NilMeter{}
+			broker.outgoingByteRate = metrics.NilMeter{}
+			broker.incomingByteRate = metrics.NilMeter{}
+			broker.requestSize = metrics.NilHistogram{}
+			broker.responseSize = metrics.NilHistogram{}
+			broker.responseRate = metrics.NilMeter{}
+			broker.requestLatency = metrics.NilHistogram{}
+			broker.requestsInFlight = metrics.NilCounter{}
 
-		conf := NewConfig()
-		conf.Net.SASL.Mechanism = SASLTypeOAuth
-		conf.Net.SASL.TokenProvider = test.tokProvider
+			conf := NewConfig()
+			conf.Net.SASL.Mechanism = SASLTypeOAuth
+			conf.Net.SASL.TokenProvider = test.tokProvider
 
-		broker.conf = conf
+			broker.conf = conf
 
-		dialer := net.Dialer{
-			Timeout:   conf.Net.DialTimeout,
-			KeepAlive: conf.Net.KeepAlive,
-			LocalAddr: conf.Net.LocalAddr,
-		}
+			dialer := net.Dialer{
+				Timeout:   conf.Net.DialTimeout,
+				KeepAlive: conf.Net.KeepAlive,
+				LocalAddr: conf.Net.LocalAddr,
+			}
 
-		conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
+			conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
 
-		if err != nil {
-			t.Fatal(err)
-		}
+			if err != nil {
+				t.Fatal(err)
+			}
 
-		broker.conn = conn
+			broker.conn = conn
 
-		err = broker.authenticateViaSASL()
+			err = broker.authenticateViaSASL()
 
-		if test.expectedBrokerError != ErrNoError {
-			if test.expectedBrokerError != err {
-				t.Errorf("[%d]:[%s] Expected %s auth error, got %s\n", i, test.name, test.expectedBrokerError, err)
+			if test.expectedBrokerError != ErrNoError {
+				if test.expectedBrokerError != err {
+					t.Errorf("[%d]:[%s] Expected %s auth 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 {
+				t.Errorf("[%d]:[%s] Unexpected error, got %s\n", i, test.name, 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 {
-			t.Errorf("[%d]:[%s] Unexpected error, got %s\n", i, test.name, err)
-		}
 
-		mockBroker.Close()
+			mockBroker.Close()
+		})
 	}
 }
 
@@ -264,7 +268,7 @@ type MockSCRAMClient struct {
 	done bool
 }
 
-func (m *MockSCRAMClient) Begin(userName, password, authzID string) (err error) {
+func (m *MockSCRAMClient) Begin(_, _, _ string) (err error) {
 	return nil
 }
 
@@ -295,7 +299,7 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 		scramChallengeResp string
 	}{
 		{
-			name:               "SASL/SCRAMSHAXXX successfull authentication",
+			name:               "SASL/SCRAMSHAXXX successful authentication",
 			mockHandshakeErr:   ErrNoError,
 			scramClient:        &MockSCRAMClient{},
 			scramChallengeResp: "pong",
@@ -325,70 +329,72 @@ 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())
-		// broker executes SASL requests against mockBroker
-		broker.requestRate = metrics.NilMeter{}
-		broker.outgoingByteRate = metrics.NilMeter{}
-		broker.incomingByteRate = metrics.NilMeter{}
-		broker.requestSize = metrics.NilHistogram{}
-		broker.responseSize = metrics.NilHistogram{}
-		broker.responseRate = metrics.NilMeter{}
-		broker.requestLatency = metrics.NilHistogram{}
-		broker.requestsInFlight = metrics.NilCounter{}
+		t.Run(test.name, func(t *testing.T) {
+			// mockBroker mocks underlying network logic and broker responses
+			mockBroker := NewMockBroker(t, 0)
+			broker := NewBroker(mockBroker.Addr())
+			// broker executes SASL requests against mockBroker
+			broker.requestRate = metrics.NilMeter{}
+			broker.outgoingByteRate = metrics.NilMeter{}
+			broker.incomingByteRate = metrics.NilMeter{}
+			broker.requestSize = metrics.NilHistogram{}
+			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})
+			mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).SetAuthBytes([]byte(test.scramChallengeResp))
+			mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).SetEnabledMechanisms([]string{SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512})
 
-		if test.mockSASLAuthErr != ErrNoError {
-			mockSASLAuthResponse = mockSASLAuthResponse.SetError(test.mockSASLAuthErr)
-		}
-		if test.mockHandshakeErr != ErrNoError {
-			mockSASLHandshakeResponse = mockSASLHandshakeResponse.SetError(test.mockHandshakeErr)
-		}
+			if test.mockSASLAuthErr != ErrNoError {
+				mockSASLAuthResponse = mockSASLAuthResponse.SetError(test.mockSASLAuthErr)
+			}
+			if test.mockHandshakeErr != ErrNoError {
+				mockSASLHandshakeResponse = mockSASLHandshakeResponse.SetError(test.mockHandshakeErr)
+			}
 
-		mockBroker.SetHandlerByMap(map[string]MockResponse{
-			"SaslAuthenticateRequest": mockSASLAuthResponse,
-			"SaslHandshakeRequest":    mockSASLHandshakeResponse,
-		})
+			mockBroker.SetHandlerByMap(map[string]MockResponse{
+				"SaslAuthenticateRequest": mockSASLAuthResponse,
+				"SaslHandshakeRequest":    mockSASLHandshakeResponse,
+			})
 
-		conf := NewConfig()
-		conf.Net.SASL.Mechanism = SASLTypeSCRAMSHA512
-		conf.Net.SASL.SCRAMClientGeneratorFunc = func() SCRAMClient { return test.scramClient }
+			conf := NewConfig()
+			conf.Net.SASL.Mechanism = SASLTypeSCRAMSHA512
+			conf.Net.SASL.SCRAMClientGeneratorFunc = func() SCRAMClient { return test.scramClient }
 
-		broker.conf = conf
-		dialer := net.Dialer{
-			Timeout:   conf.Net.DialTimeout,
-			KeepAlive: conf.Net.KeepAlive,
-			LocalAddr: conf.Net.LocalAddr,
-		}
+			broker.conf = conf
+			dialer := net.Dialer{
+				Timeout:   conf.Net.DialTimeout,
+				KeepAlive: conf.Net.KeepAlive,
+				LocalAddr: conf.Net.LocalAddr,
+			}
 
-		conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
+			conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
 
-		if err != nil {
-			t.Fatal(err)
-		}
+			if err != nil {
+				t.Fatal(err)
+			}
 
-		broker.conn = conn
+			broker.conn = conn
 
-		err = broker.authenticateViaSASL()
+			err = broker.authenticateViaSASL()
 
-		if test.mockSASLAuthErr != ErrNoError {
-			if test.mockSASLAuthErr != err {
-				t.Errorf("[%d]:[%s] Expected %s SASL authentication error, got %s\n", i, test.name, test.mockHandshakeErr, err)
-			}
-		} else if test.mockHandshakeErr != ErrNoError {
-			if test.mockHandshakeErr != err {
-				t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.mockHandshakeErr, err)
+			if test.mockSASLAuthErr != ErrNoError {
+				if test.mockSASLAuthErr != err {
+					t.Errorf("[%d]:[%s] Expected %s SASL authentication error, got %s\n", i, test.name, test.mockHandshakeErr, err)
+				}
+			} else if test.mockHandshakeErr != ErrNoError {
+				if test.mockHandshakeErr != err {
+					t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.mockHandshakeErr, 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 {
+				t.Errorf("[%d]:[%s] Unexpected error, got %s\n", i, test.name, 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 {
-			t.Errorf("[%d]:[%s] Unexpected error, got %s\n", i, test.name, err)
-		}
 
-		mockBroker.Close()
+			mockBroker.Close()
+		})
 	}
 }
 
@@ -424,96 +430,98 @@ func TestSASLPlainAuth(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-		// mockBroker mocks underlying network logic and broker responses
-		mockBroker := NewMockBroker(t, 0)
+		t.Run(test.name, func(t *testing.T) {
+			// mockBroker mocks underlying network logic and broker responses
+			mockBroker := NewMockBroker(t, 0)
 
-		mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).
-			SetAuthBytes([]byte(`response_payload`))
+			mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).
+				SetAuthBytes([]byte(`response_payload`))
 
-		if test.mockAuthErr != ErrNoError {
-			mockSASLAuthResponse = mockSASLAuthResponse.SetError(test.mockAuthErr)
-		}
+			if test.mockAuthErr != ErrNoError {
+				mockSASLAuthResponse = mockSASLAuthResponse.SetError(test.mockAuthErr)
+			}
 
-		mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).
-			SetEnabledMechanisms([]string{SASLTypePlaintext})
+			mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).
+				SetEnabledMechanisms([]string{SASLTypePlaintext})
 
-		if test.mockHandshakeErr != ErrNoError {
-			mockSASLHandshakeResponse = mockSASLHandshakeResponse.SetError(test.mockHandshakeErr)
-		}
+			if test.mockHandshakeErr != ErrNoError {
+				mockSASLHandshakeResponse = mockSASLHandshakeResponse.SetError(test.mockHandshakeErr)
+			}
 
-		mockBroker.SetHandlerByMap(map[string]MockResponse{
-			"SaslAuthenticateRequest": mockSASLAuthResponse,
-			"SaslHandshakeRequest":    mockSASLHandshakeResponse,
-		})
+			mockBroker.SetHandlerByMap(map[string]MockResponse{
+				"SaslAuthenticateRequest": mockSASLAuthResponse,
+				"SaslHandshakeRequest":    mockSASLHandshakeResponse,
+			})
 
-		// broker executes SASL requests against mockBroker
-		broker := NewBroker(mockBroker.Addr())
-		broker.requestRate = metrics.NilMeter{}
-		broker.outgoingByteRate = metrics.NilMeter{}
-		broker.incomingByteRate = metrics.NilMeter{}
-		broker.requestSize = metrics.NilHistogram{}
-		broker.responseSize = metrics.NilHistogram{}
-		broker.responseRate = metrics.NilMeter{}
-		broker.requestLatency = metrics.NilHistogram{}
-		broker.requestsInFlight = metrics.NilCounter{}
+			// broker executes SASL requests against mockBroker
+			broker := NewBroker(mockBroker.Addr())
+			broker.requestRate = metrics.NilMeter{}
+			broker.outgoingByteRate = metrics.NilMeter{}
+			broker.incomingByteRate = metrics.NilMeter{}
+			broker.requestSize = metrics.NilHistogram{}
+			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
+			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
 
-		broker.conf = conf
-		broker.conf.Version = V1_0_0_0
-		dialer := net.Dialer{
-			Timeout:   conf.Net.DialTimeout,
-			KeepAlive: conf.Net.KeepAlive,
-			LocalAddr: conf.Net.LocalAddr,
-		}
-
-		conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
-
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		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])
+			broker.conf = conf
+			broker.conf.Version = V1_0_0_0
+			dialer := net.Dialer{
+				Timeout:   conf.Net.DialTimeout,
+				KeepAlive: conf.Net.KeepAlive,
+				LocalAddr: conf.Net.LocalAddr,
+			}
+
+			conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
+
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			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 {
-				t.Errorf("[%d]:[%s] Expected %s auth error, got %s\n", i, test.name, test.mockAuthErr, err)
-			}
-		} else if test.mockHandshakeErr != ErrNoError {
-			if test.mockHandshakeErr != err {
-				t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.mockHandshakeErr, err)
+			if test.mockAuthErr != ErrNoError {
+				if test.mockAuthErr != err {
+					t.Errorf("[%d]:[%s] Expected %s auth error, got %s\n", i, test.name, test.mockAuthErr, err)
+				}
+			} else if test.mockHandshakeErr != ErrNoError {
+				if test.mockHandshakeErr != err {
+					t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.mockHandshakeErr, 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 {
+				t.Errorf("[%d]:[%s] Unexpected error, got %s\n", i, test.name, 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 {
-			t.Errorf("[%d]:[%s] Unexpected error, got %s\n", i, test.name, err)
-		}
 
-		mockBroker.Close()
+			mockBroker.Close()
+		})
 	}
 }
 
@@ -616,73 +624,75 @@ func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
 		},
 	}
 	for i, test := range testTable {
-		mockBroker := NewMockBroker(t, 0)
-		// broker executes SASL requests against mockBroker
+		t.Run(test.name, func(t *testing.T) {
+			mockBroker := NewMockBroker(t, 0)
+			// broker executes SASL requests against mockBroker
+
+			mockBroker.SetGSSAPIHandler(func(bytes []byte) []byte {
+				return nil
+			})
+			broker := NewBroker(mockBroker.Addr())
+			broker.requestRate = metrics.NilMeter{}
+			broker.outgoingByteRate = metrics.NilMeter{}
+			broker.incomingByteRate = metrics.NilMeter{}
+			broker.requestSize = metrics.NilHistogram{}
+			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"
+			conf.Net.SASL.GSSAPI.KerberosConfigPath = "krb5.conf"
+			conf.Net.SASL.GSSAPI.Realm = "EXAMPLE.COM"
+			conf.Net.SASL.GSSAPI.Username = "kafka"
+			conf.Net.SASL.GSSAPI.Password = "kafka"
+			conf.Net.SASL.GSSAPI.KeyTabPath = "kafka.keytab"
+			conf.Net.SASL.GSSAPI.AuthType = KRB5_USER_AUTH
+			broker.conf = conf
+			broker.conf.Version = V1_0_0_0
+			dialer := net.Dialer{
+				Timeout:   conf.Net.DialTimeout,
+				KeepAlive: conf.Net.KeepAlive,
+				LocalAddr: conf.Net.LocalAddr,
+			}
+
+			conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
 
-		mockBroker.SetGSSAPIHandler(func(bytes []byte) []byte {
-			return nil
-		})
-		broker := NewBroker(mockBroker.Addr())
-		broker.requestRate = metrics.NilMeter{}
-		broker.outgoingByteRate = metrics.NilMeter{}
-		broker.incomingByteRate = metrics.NilMeter{}
-		broker.requestSize = metrics.NilHistogram{}
-		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"
-		conf.Net.SASL.GSSAPI.KerberosConfigPath = "krb5.conf"
-		conf.Net.SASL.GSSAPI.Realm = "EXAMPLE.COM"
-		conf.Net.SASL.GSSAPI.Username = "kafka"
-		conf.Net.SASL.GSSAPI.Password = "kafka"
-		conf.Net.SASL.GSSAPI.KeyTabPath = "kafka.keytab"
-		conf.Net.SASL.GSSAPI.AuthType = KRB5_USER_AUTH
-		broker.conf = conf
-		broker.conf.Version = V1_0_0_0
-		dialer := net.Dialer{
-			Timeout:   conf.Net.DialTimeout,
-			KeepAlive: conf.Net.KeepAlive,
-			LocalAddr: conf.Net.LocalAddr,
-		}
-
-		conn, err := dialer.Dial("tcp", mockBroker.listener.Addr().String())
-
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		gssapiHandler := KafkaGSSAPIHandler{
-			client:         &MockKerberosClient{},
-			badResponse:    test.badResponse,
-			badKeyChecksum: test.badKeyChecksum,
-		}
-		mockBroker.SetGSSAPIHandler(gssapiHandler.MockKafkaGSSAPI)
-		broker.conn = conn
-		if test.mockKerberosClient {
-			broker.kerberosAuthenticator.NewKerberosClientFunc = func(config *GSSAPIConfig) (KerberosClient, error) {
-				return &MockKerberosClient{
-					mockError:  test.error,
-					errorStage: test.errorStage,
-				}, nil
-			}
-		} else {
-			broker.kerberosAuthenticator.NewKerberosClientFunc = nil
-		}
-
-		err = broker.authenticateViaSASL()
-
-		if err != nil && test.error != nil {
-			if test.error.Error() != err.Error() {
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			gssapiHandler := KafkaGSSAPIHandler{
+				client:         &MockKerberosClient{},
+				badResponse:    test.badResponse,
+				badKeyChecksum: test.badKeyChecksum,
+			}
+			mockBroker.SetGSSAPIHandler(gssapiHandler.MockKafkaGSSAPI)
+			broker.conn = conn
+			if test.mockKerberosClient {
+				broker.kerberosAuthenticator.NewKerberosClientFunc = func(config *GSSAPIConfig) (KerberosClient, error) {
+					return &MockKerberosClient{
+						mockError:  test.error,
+						errorStage: test.errorStage,
+					}, nil
+				}
+			} else {
+				broker.kerberosAuthenticator.NewKerberosClientFunc = nil
+			}
+
+			err = broker.authenticateViaSASL()
+
+			if err != nil && test.error != nil {
+				if test.error.Error() != err.Error() {
+					t.Errorf("[%d] Expected error:%s, got:%s.", i, test.error, err)
+				}
+			} else if (err == nil && test.error != nil) || (err != nil && test.error == nil) {
 				t.Errorf("[%d] Expected error:%s, got:%s.", i, test.error, err)
 			}
-		} else if (err == nil && test.error != nil) || (err != nil && test.error == nil) {
-			t.Errorf("[%d] Expected error:%s, got:%s.", i, test.error, err)
-		}
 
-		mockBroker.Close()
+			mockBroker.Close()
+		})
 	}
 }
 
@@ -723,17 +733,19 @@ func TestBuildClientFirstMessage(t *testing.T) {
 	}
 
 	for i, test := range testTable {
-		actual, err := buildClientFirstMessage(test.token)
-
-		if !reflect.DeepEqual(test.expected, actual) {
-			t.Errorf("Expected %s, got %s\n", test.expected, actual)
-		}
-		if test.expectError && err == nil {
-			t.Errorf("[%d]:[%s] Expected an error but did not get one", i, test.name)
-		}
-		if !test.expectError && err != nil {
-			t.Errorf("[%d]:[%s] Expected no error but got %s\n", i, test.name, err)
-		}
+		t.Run(test.name, func(t *testing.T) {
+			actual, err := buildClientFirstMessage(test.token)
+
+			if !reflect.DeepEqual(test.expected, actual) {
+				t.Errorf("Expected %s, got %s\n", test.expected, actual)
+			}
+			if test.expectError && err == nil {
+				t.Errorf("[%d]:[%s] Expected an error but did not get one", i, test.name)
+			}
+			if !test.expectError && err != nil {
+				t.Errorf("[%d]:[%s] Expected no error but got %s\n", i, test.name, err)
+			}
+		})
 	}
 }
 
diff --git a/client_test.go b/client_test.go
index cd47a50..9eee454 100644
--- a/client_test.go
+++ b/client_test.go
@@ -642,7 +642,7 @@ func TestClientController(t *testing.T) {
 	}
 	defer safeClose(t, client2)
 	if _, err = client2.Controller(); err != ErrUnsupportedVersion {
-		t.Errorf("Expected Contoller() to return %s, found %s", ErrUnsupportedVersion, err)
+		t.Errorf("Expected Controller() to return %s, found %s", ErrUnsupportedVersion, err)
 	}
 }
 
diff --git a/client_tls_test.go b/client_tls_test.go
index 0e47e17..e366127 100644
--- a/client_tls_test.go
+++ b/client_tls_test.go
@@ -1,16 +1,15 @@
 package sarama
 
 import (
-	"math/big"
-	"net"
-	"testing"
-	"time"
-
 	"crypto/rand"
 	"crypto/rsa"
 	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
+	"math/big"
+	"net"
+	"testing"
+	"time"
 )
 
 func TestTLS(t *testing.T) {
@@ -95,10 +94,12 @@ func TestTLS(t *testing.T) {
 	}
 
 	for _, tc := range []struct {
+		name           string
 		Succeed        bool
 		Server, Client *tls.Config
 	}{
-		{ // Verify client fails if wrong CA cert pool is specified
+		{
+			name:    "Verify client fails if wrong CA cert pool is specified",
 			Succeed: false,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
@@ -109,7 +110,8 @@ func TestTLS(t *testing.T) {
 				}},
 			},
 		},
-		{ // Verify client fails if wrong key is specified
+		{
+			name:    "Verify client fails if wrong key is specified",
 			Succeed: false,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
@@ -120,7 +122,8 @@ func TestTLS(t *testing.T) {
 				}},
 			},
 		},
-		{ // Verify client fails if wrong cert is specified
+		{
+			name:    "Verify client fails if wrong cert is specified",
 			Succeed: false,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
@@ -131,7 +134,8 @@ func TestTLS(t *testing.T) {
 				}},
 			},
 		},
-		{ // Verify client fails if no CAs are specified
+		{
+			name:    "Verify client fails if no CAs are specified",
 			Succeed: false,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
@@ -141,18 +145,21 @@ func TestTLS(t *testing.T) {
 				}},
 			},
 		},
-		{ // Verify client fails if no keys are specified
+		{
+			name:    "Verify client fails if no keys are specified",
 			Succeed: false,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
 				RootCAs: pool,
 			},
 		},
-		{ // Finally, verify it all works happily with client and server cert in place
+		{
+			name:    "Finally, verify it all works happily with client and server cert in place",
 			Succeed: true,
 			Server:  serverTLSConfig,
 			Client: &tls.Config{
-				RootCAs: pool,
+				RootCAs:    pool,
+				ServerName: "127.0.0.1",
 				Certificates: []tls.Certificate{{
 					Certificate: [][]byte{clientDer},
 					PrivateKey:  clientkey,
@@ -160,7 +167,7 @@ func TestTLS(t *testing.T) {
 			},
 		},
 	} {
-		doListenerTLSTest(t, tc.Succeed, tc.Server, tc.Client)
+		t.Run(tc.name, func(t *testing.T) { doListenerTLSTest(t, tc.Succeed, tc.Server, tc.Client) })
 	}
 }
 
diff --git a/functional_producer_test.go b/functional_producer_test.go
index 1fa0ba1..e589a8e 100644
--- a/functional_producer_test.go
+++ b/functional_producer_test.go
@@ -3,6 +3,7 @@ package sarama
 import (
 	"fmt"
 	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -96,6 +97,83 @@ func TestFuncProducingToInvalidTopic(t *testing.T) {
 	safeClose(t, producer)
 }
 
+func TestFuncProducingIdempotentWithBrokerFailure(t *testing.T) {
+	setupFunctionalTest(t)
+	defer teardownFunctionalTest(t)
+
+	config := NewConfig()
+	config.Producer.Flush.Frequency = 250 * time.Millisecond
+	config.Producer.Idempotent = true
+	config.Producer.Timeout = 500 * time.Millisecond
+	config.Producer.Retry.Max = 1
+	config.Producer.Retry.Backoff = 500 * time.Millisecond
+	config.Producer.Return.Successes = true
+	config.Producer.Return.Errors = true
+	config.Producer.RequiredAcks = WaitForAll
+	config.Net.MaxOpenRequests = 1
+	config.Version = V0_11_0_0
+
+	producer, err := NewSyncProducer(kafkaBrokers, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer safeClose(t, producer)
+
+	// Successfully publish a few messages
+	for i := 0; i < 10; i++ {
+		_, _, err = producer.SendMessage(&ProducerMessage{
+			Topic: "test.1",
+			Value: StringEncoder(fmt.Sprintf("%d message", i)),
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// break the brokers.
+	for proxyName, proxy := range Proxies {
+		if !strings.Contains(proxyName, "kafka") {
+			continue
+		}
+		if err := proxy.Disable(); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// This should fail hard now
+	for i := 10; i < 20; i++ {
+		_, _, err = producer.SendMessage(&ProducerMessage{
+			Topic: "test.1",
+			Value: StringEncoder(fmt.Sprintf("%d message", i)),
+		})
+		if err == nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Now bring the proxy back up
+	for proxyName, proxy := range Proxies {
+		if !strings.Contains(proxyName, "kafka") {
+			continue
+		}
+		if err := proxy.Enable(); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// We should be able to publish again (once everything calms down)
+	// (otherwise it times out)
+	for {
+		_, _, err = producer.SendMessage(&ProducerMessage{
+			Topic: "test.1",
+			Value: StringEncoder("comeback message"),
+		})
+		if err == nil {
+			break
+		}
+	}
+}
+
 func testProducingMessages(t *testing.T, config *Config) {
 	setupFunctionalTest(t)
 	defer teardownFunctionalTest(t)
diff --git a/list_partition_reassignments_request_test.go b/list_partition_reassignments_request_test.go
index d9f9f92..41a5b9c 100644
--- a/list_partition_reassignments_request_test.go
+++ b/list_partition_reassignments_request_test.go
@@ -4,7 +4,7 @@ import "testing"
 
 var (
 	listPartitionReassignmentsRequestOneBlock = []byte{
-		0, 0, 39, 16, // timout 10000
+		0, 0, 39, 16, // timeout 10000
 		2,                         // 2-1=1 block
 		6, 116, 111, 112, 105, 99, // topic name "topic" as compact string
 		2,          // 2-1=1 partitions
diff --git a/record_test.go b/record_test.go
index 2756c5b..1aceeda 100644
--- a/record_test.go
+++ b/record_test.go
@@ -283,7 +283,7 @@ func TestRecordBatchDecoding(t *testing.T) {
 			r.length = varintLengthField{}
 		}
 		// The compression level is not restored on decoding. It is not needed
-		// anyway. We only set it here to ensure that comparision succeeds.
+		// anyway. We only set it here to ensure that comparison succeeds.
 		batch.CompressionLevel = tc.batch.CompressionLevel
 		if !reflect.DeepEqual(batch, tc.batch) {
 			t.Errorf(spew.Sprintf("invalid decode of %s\ngot %+v\nwanted %+v", tc.name, batch, tc.batch))
diff --git a/sasl_authenticate_response_test.go b/sasl_authenticate_response_test.go
index 0444796..c555cfb 100644
--- a/sasl_authenticate_response_test.go
+++ b/sasl_authenticate_response_test.go
@@ -17,5 +17,5 @@ func TestSaslAuthenticateResponse(t *testing.T) {
 	response.ErrorMessage = &msg
 	response.SaslAuthBytes = []byte(`msg`)
 
-	testResponse(t, "authenticate reponse", response, saslAuthenticatResponseErr)
+	testResponse(t, "authenticate response", response, saslAuthenticatResponseErr)
 }