Repository: Shopify/sarama
Pull request #2131 (2022-02-07 15:29:06)

feat: make `ErrOutOfBrokers` wrap the underlying error that prevented connections to the brokers
With this change, the user of Sarama is required to use Go 1.13's `errors.Is` etc (rather then `==`) when forming conditionals returned by this library.

bai: @k-wall Thanks for your contribution! Looks like there are some linter failures on Go 1.17, could you please take a look?
k-wall: @bai thanks for looking.  

I'm not certain what the output from the Github Action is trying to tell me.  I think you are referring to these errors in the output. 

```
run golangci-lint
  Running [/home/runner/golangci-lint-1.44.0-linux-amd64/golangci-lint run --out-format=github-actions] in [] ...
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
  Error: undeclared name: `errors` (typecheck)
```

However, when I run the tooling locally against my branch, I see a different set of errors.

```
% go version
go version go1.17.6 darwin/amd64
% golangci-lint version
golangci-lint has version 1.44.0 built from 617470f on 2022-01-25T11:22:07Z
``

```
golangci-lint run --out-format=github-actions
::error file=metrics_test.go,line=103,col=6::`minCountMeterValidator` is unused (deadcode)
::error file=metrics_test.go,line=134,col=6::`minCountHistogramValidator` is unused (deadcode)
::error file=metrics_test.go,line=156,col=6::`minValHistogramValidator` is unused (deadcode)
::error file=metrics_test.go,line=165,col=6::`maxValHistogramValidator` is unused (deadcode)
::error file=metrics_test.go,line=60,col=28::func `(*metricValidators).registerForGlobalAndTopic` is unused (unused)
```

Can you give me a pointer to help diagnose the problem?

dnwe: @k-wall the functional tests are gated behind a build flag so to lint the full source code directly you need to set that flag before running golangci-lint (or you can use the makefile version which does it for you via `make lint`):

```
GOFLAGS="-tags=functional" golangci-lint
functional_client_test.go:23:6: undeclared name: `errors` (typecheck)
        if !errors.Is(err, ErrOutOfBrokers) {
            ^
functional_client_test.go:40:54: undeclared name: `errors` (typecheck)
        if err := client.RefreshMetadata("unknown_topic"); !errors.Is(err, ErrUnknownTopicOrPartition) {
                                                            ^
functional_client_test.go:44:51: undeclared name: `errors` (typecheck)
        if _, err := client.Leader("unknown_topic", 0); !errors.Is(err, ErrUnknownTopicOrPartition) {
                                                         ^
functional_client_test.go:48:53: undeclared name: `errors` (typecheck)
        if _, err := client.Replicas("invalid/topic", 0); !errors.Is(err, ErrUnknownTopicOrPartition) && !errors.Is(err, ErrInvalidTopic) {
                                                           ^
functional_consumer_group_test.go:532:40: undeclared name: `errors` (typecheck)
                if err := m.Consume(ctx, topics, m); errors.Is(err, ErrClosedConsumerGroup) {
                                                     ^
functional_consumer_test.go:27:61: undeclared name: `errors` (typecheck)
        if _, err := consumer.ConsumePartition("test.1", 0, -10); !errors.Is(err, ErrOffsetOutOfRange) {
                                                                   ^
functional_consumer_test.go:31:71: undeclared name: `errors` (typecheck)
        if _, err := consumer.ConsumePartition("test.1", 0, math.MaxInt64); !errors.Is(err, ErrOffsetOutOfRange) {
                                                                             ^
functional_producer_test.go:100:78: undeclared name: `errors` (typecheck)
        if _, _, err := producer.SendMessage(&ProducerMessage{Topic: "in/valid"}); !errors.Is(err, ErrUnknownTopicOrPartition) && !errors.Is(err, ErrInvalidTopic) {
                                                                                    ^
functional_test.go:388:9: undeclared name: `errors` (typecheck)
        return errors.Is(err, ErrUnknownTopicOrPartition) || errors.Is(err, ErrInvalidTopic) || errors.Is(err, ErrNoError)
               ^
functional_test.go:392:9: undeclared name: `errors` (typecheck)
        return errors.Is(err, ErrTopicAlreadyExists) || errors.Is(err, ErrNoError)
```

These files are just missing the "errors" import entirely — you can fix them with `GOFLAGS="-tags=functional" goimports -w *.go`
k-wall: Thanks for the feedback, I'll update the PR based in the feedback.

I'd also like to ask about the behaviour of `Client#InitProducerID()`.  I think it behaves in a different way to the other parts of the API that return `ErrOutOfBrokers` (`RefreshMetadata` etc).  If I understand its current behaviour, in the event all brokers becoming unconnectable (say) it will return the error returned by the last broker tried.   I think the only time it can return a `ErrOutOfBrokers` result is if `client.any()` returns nil on first attempt (which I'm not sure can ever actually happen). 

Is there a reason why it error contract is different?   I am wondering if it should actually return ErrOutOfBrokers, and with this change, wrap the errors from the brokers?
dnwe: @k-wall yes I agree, it would make sense for the overall error `ErrOutOfBrokers` to be returned from InitProducerID too and the per-broker errors wrapped 

Patch:
diff --git a/.gitignore b/.gitignore
index 2c9adc2..2384330 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,3 +27,5 @@ coverage.txt
 profile.out
 
 simplest-uncommitted-msg-0.1-jar-with-dependencies.jar
+
+.idea
diff --git a/.golangci.yml b/.golangci.yml
index c1e71ac..3859c00 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -44,6 +44,7 @@ linters:
     - dogsled
     # - dupl
     - errcheck
+    - errorlint
     - funlen
     - gochecknoinits
     # - goconst
diff --git a/admin.go b/admin.go
index 367e124..e3a81c2 100644
--- a/admin.go
+++ b/admin.go
@@ -183,15 +183,7 @@ func (ca *clusterAdmin) refreshController() (*Broker, error) {
 // isErrNoController returns `true` if the given error type unwraps to an
 // `ErrNotController` response from Kafka
 func isErrNoController(err error) bool {
-	switch e := err.(type) {
-	case *TopicError:
-		return e.Err == ErrNotController
-	case *TopicPartitionError:
-		return e.Err == ErrNotController
-	case KError:
-		return e == ErrNotController
-	}
-	return false
+	return errors.Is(err, ErrNotController)
 }
 
 // retryOnError will repeatedly call the given (error-returning) func in the
@@ -255,8 +247,8 @@ func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateO
 			return ErrIncompleteResponse
 		}
 
-		if topicErr.Err != ErrNoError {
-			if topicErr.Err == ErrNotController {
+		if !errors.Is(topicErr.Err, ErrNoError) {
+			if errors.Is(topicErr.Err, ErrNotController) {
 				_, _ = ca.refreshController()
 			}
 			return topicErr
@@ -442,8 +434,8 @@ func (ca *clusterAdmin) DeleteTopic(topic string) error {
 			return ErrIncompleteResponse
 		}
 
-		if topicErr != ErrNoError {
-			if topicErr == ErrNotController {
+		if !errors.Is(topicErr, ErrNoError) {
+			if errors.Is(topicErr, ErrNotController) {
 				_, _ = ca.refreshController()
 			}
 			return topicErr
@@ -483,8 +475,8 @@ func (ca *clusterAdmin) CreatePartitions(topic string, count int32, assignment [
 			return ErrIncompleteResponse
 		}
 
-		if topicErr.Err != ErrNoError {
-			if topicErr.Err == ErrNotController {
+		if !errors.Is(topicErr.Err, ErrNoError) {
+			if errors.Is(topicErr.Err, ErrNotController) {
 				_, _ = ca.refreshController()
 			}
 			return topicErr
@@ -527,7 +519,7 @@ func (ca *clusterAdmin) AlterPartitionReassignments(topic string, assignment [][
 
 			for topic, topicErrors := range rsp.Errors {
 				for partition, partitionError := range topicErrors {
-					if partitionError.errorCode != ErrNoError {
+					if !errors.Is(partitionError.errorCode, ErrNoError) {
 						errStr := fmt.Sprintf("[%s-%d]: %s", topic, partition, partitionError.errorCode.Error())
 						errs = append(errs, errors.New(errStr))
 					}
@@ -536,7 +528,7 @@ func (ca *clusterAdmin) AlterPartitionReassignments(topic string, assignment [][
 		}
 
 		if len(errs) > 0 {
-			return ErrReassignPartitions{MultiError{&errs}}
+			return Wrap(ErrReassignPartitions, errs...)
 		}
 
 		return nil
@@ -604,7 +596,7 @@ func (ca *clusterAdmin) DeleteRecords(topic string, partitionOffsets map[int32]i
 				errs = append(errs, ErrIncompleteResponse)
 			} else {
 				for _, deleteRecordsResponsePartition := range deleteRecordsResponseTopic.Partitions {
-					if deleteRecordsResponsePartition.Err != ErrNoError {
+					if !errors.Is(deleteRecordsResponsePartition.Err, ErrNoError) {
 						errs = append(errs, errors.New(deleteRecordsResponsePartition.Err.Error()))
 					}
 				}
@@ -612,7 +604,7 @@ func (ca *clusterAdmin) DeleteRecords(topic string, partitionOffsets map[int32]i
 		}
 	}
 	if len(errs) > 0 {
-		return ErrDeleteRecords{MultiError{&errs}}
+		return Wrap(ErrDeleteRecords, errs...)
 	}
 	// todo since we are dealing with couple of partitions it would be good if we return slice of errors
 	// for each partition instead of one error
@@ -966,11 +958,11 @@ func (ca *clusterAdmin) DeleteConsumerGroupOffset(group string, topic string, pa
 		return err
 	}
 
-	if resp.ErrorCode != ErrNoError {
+	if !errors.Is(resp.ErrorCode, ErrNoError) {
 		return resp.ErrorCode
 	}
 
-	if resp.Errors[topic][partition] != ErrNoError {
+	if !errors.Is(resp.Errors[topic][partition], ErrNoError) {
 		return resp.Errors[topic][partition]
 	}
 	return nil
@@ -996,7 +988,7 @@ func (ca *clusterAdmin) DeleteConsumerGroup(group string) error {
 		return ErrIncompleteResponse
 	}
 
-	if groupErr != ErrNoError {
+	if !errors.Is(groupErr, ErrNoError) {
 		return groupErr
 	}
 
@@ -1128,7 +1120,7 @@ func (ca *clusterAdmin) DescribeClientQuotas(components []QuotaFilterComponent,
 	if rsp.ErrorMsg != nil && len(*rsp.ErrorMsg) > 0 {
 		return nil, errors.New(*rsp.ErrorMsg)
 	}
-	if rsp.ErrorCode != ErrNoError {
+	if !errors.Is(rsp.ErrorCode, ErrNoError) {
 		return nil, rsp.ErrorCode
 	}
 
@@ -1160,7 +1152,7 @@ func (ca *clusterAdmin) AlterClientQuotas(entity []QuotaEntityComponent, op Clie
 		if entry.ErrorMsg != nil && len(*entry.ErrorMsg) > 0 {
 			return errors.New(*entry.ErrorMsg)
 		}
-		if entry.ErrorCode != ErrNoError {
+		if !errors.Is(entry.ErrorCode, ErrNoError) {
 			return entry.ErrorCode
 		}
 	}
diff --git a/async_producer.go b/async_producer.go
index c0dcce9..27398bc 100644
--- a/async_producer.go
+++ b/async_producer.go
@@ -2,6 +2,7 @@ package sarama
 
 import (
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -1058,12 +1059,12 @@ func (p *asyncProducer) retryBatch(topic string, partition int32, pSet *partitio
 }
 
 func (bp *brokerProducer) handleError(sent *produceSet, err error) {
-	switch err.(type) {
-	case PacketEncodingError:
+	var target PacketEncodingError
+	if errors.As(err, &target) {
 		sent.eachPartition(func(topic string, partition int32, pSet *partitionSet) {
 			bp.parent.returnErrors(pSet.msgs, err)
 		})
-	default:
+	} else {
 		Logger.Printf("producer/broker/%d state change to [closing] because %s\n", bp.broker.ID(), err)
 		bp.parent.abandonBrokerConnection(bp.broker)
 		_ = bp.broker.Close()
diff --git a/broker.go b/broker.go
index c60e9a0..4b3ae84 100644
--- a/broker.go
+++ b/broker.go
@@ -3,6 +3,7 @@ package sarama
 import (
 	"crypto/tls"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -1161,7 +1162,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 		return err
 	}
 
-	if res.Err != ErrNoError {
+	if !errors.Is(res.Err, ErrNoError) {
 		Logger.Printf("Invalid SASL Mechanism : %s\n", res.Err.Error())
 		return res.Err
 	}
@@ -1351,12 +1352,12 @@ func (b *Broker) sendAndReceiveSASLSCRAMv0() error {
 
 	scramClient := b.conf.Net.SASL.SCRAMClientGeneratorFunc()
 	if err := scramClient.Begin(b.conf.Net.SASL.User, b.conf.Net.SASL.Password, b.conf.Net.SASL.SCRAMAuthzID); err != nil {
-		return fmt.Errorf("failed to start SCRAM exchange with the server: %s", err.Error())
+		return fmt.Errorf("failed to start SCRAM exchange with the server: %w", err)
 	}
 
 	msg, err := scramClient.Step("")
 	if err != nil {
-		return fmt.Errorf("failed to advance the SCRAM exchange: %s", err.Error())
+		return fmt.Errorf("failed to advance the SCRAM exchange: %w", err)
 	}
 
 	for !scramClient.Done() {
@@ -1408,12 +1409,12 @@ func (b *Broker) sendAndReceiveSASLSCRAMv1() error {
 
 	scramClient := b.conf.Net.SASL.SCRAMClientGeneratorFunc()
 	if err := scramClient.Begin(b.conf.Net.SASL.User, b.conf.Net.SASL.Password, b.conf.Net.SASL.SCRAMAuthzID); err != nil {
-		return fmt.Errorf("failed to start SCRAM exchange with the server: %s", err.Error())
+		return fmt.Errorf("failed to start SCRAM exchange with the server: %w", err)
 	}
 
 	msg, err := scramClient.Step("")
 	if err != nil {
-		return fmt.Errorf("failed to advance the SCRAM exchange: %s", err.Error())
+		return fmt.Errorf("failed to advance the SCRAM exchange: %w", err)
 	}
 
 	for !scramClient.Done() {
@@ -1487,7 +1488,7 @@ func (b *Broker) receiveSaslAuthenticateResponse(correlationID int32) ([]byte, e
 	if err := versionedDecode(buf, res, 0); err != nil {
 		return nil, err
 	}
-	if res.Err != ErrNoError {
+	if !errors.Is(res.Err, ErrNoError) {
 		return nil, res.Err
 	}
 	return res.SaslAuthBytes, nil
@@ -1577,7 +1578,7 @@ func (b *Broker) receiveSASLServerResponse(res *SaslAuthenticateResponse, correl
 		return bytesRead, err
 	}
 
-	if res.Err != ErrNoError {
+	if !errors.Is(res.Err, ErrNoError) {
 		return bytesRead, res.Err
 	}
 
diff --git a/client.go b/client.go
index 1c69cb8..8f07e42 100644
--- a/client.go
+++ b/client.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"math/rand"
 	"sort"
 	"sync"
@@ -168,13 +169,11 @@ func NewClient(addrs []string, conf *Config) (Client, error) {
 	if conf.Metadata.Full {
 		// do an initial fetch of all cluster metadata by specifying an empty list of topics
 		err := client.RefreshMetadata()
-		switch err {
-		case nil:
-			break
-		case ErrLeaderNotAvailable, ErrReplicaNotAvailable, ErrTopicAuthorizationFailed, ErrClusterAuthorizationFailed:
+		if err == nil {
+		} else if errors.Is(err, ErrLeaderNotAvailable) || errors.Is(err, ErrReplicaNotAvailable) || errors.Is(err, ErrTopicAuthorizationFailed) || errors.Is(err, ErrClusterAuthorizationFailed) {
 			// indicates that maybe part of the cluster is down, but is not fatal to creating the client
 			Logger.Println(err)
-		default:
+		} else {
 			close(client.closed) // we haven't started the background updater yet, so we have to do this manually
 			_ = client.Close()
 			return nil, err
@@ -213,24 +212,24 @@ func (client *client) Broker(brokerID int32) (*Broker, error) {
 }
 
 func (client *client) InitProducerID() (*InitProducerIDResponse, error) {
-	err := ErrOutOfBrokers
+	brokerErrors := make([]error, 0)
 	for broker := client.any(); broker != nil; broker = client.any() {
 		var response *InitProducerIDResponse
 		req := &InitProducerIDRequest{}
 
-		response, err = broker.InitProducerID(req)
-		switch err.(type) {
-		case nil:
+		response, err := broker.InitProducerID(req)
+		if err == nil {
 			return response, nil
-		default:
+		} else {
 			// some error, remove that broker and try again
 			Logger.Printf("Client got error from broker %d when issuing InitProducerID : %v\n", broker.ID(), err)
 			_ = broker.Close()
+			brokerErrors = append(brokerErrors, err)
 			client.deregisterBroker(broker)
 		}
 	}
 
-	return nil, err
+	return nil, Wrap(ErrOutOfBrokers, brokerErrors...)
 }
 
 func (client *client) Close() error {
@@ -373,7 +372,7 @@ func (client *client) Replicas(topic string, partitionID int32) ([]int32, error)
 		return nil, ErrUnknownTopicOrPartition
 	}
 
-	if metadata.Err == ErrReplicaNotAvailable {
+	if errors.Is(metadata.Err, ErrReplicaNotAvailable) {
 		return dupInt32Slice(metadata.Replicas), metadata.Err
 	}
 	return dupInt32Slice(metadata.Replicas), nil
@@ -398,7 +397,7 @@ func (client *client) InSyncReplicas(topic string, partitionID int32) ([]int32,
 		return nil, ErrUnknownTopicOrPartition
 	}
 
-	if metadata.Err == ErrReplicaNotAvailable {
+	if errors.Is(metadata.Err, ErrReplicaNotAvailable) {
 		return dupInt32Slice(metadata.Isr), metadata.Err
 	}
 	return dupInt32Slice(metadata.Isr), nil
@@ -423,7 +422,7 @@ func (client *client) OfflineReplicas(topic string, partitionID int32) ([]int32,
 		return nil, ErrUnknownTopicOrPartition
 	}
 
-	if metadata.Err == ErrReplicaNotAvailable {
+	if errors.Is(metadata.Err, ErrReplicaNotAvailable) {
 		return dupInt32Slice(metadata.OfflineReplicas), metadata.Err
 	}
 	return dupInt32Slice(metadata.OfflineReplicas), nil
@@ -742,7 +741,7 @@ func (client *client) setPartitionCache(topic string, partitionSet partitionType
 
 	ret := make([]int32, 0, len(partitions))
 	for _, partition := range partitions {
-		if partitionSet == writablePartitions && partition.Err == ErrLeaderNotAvailable {
+		if partitionSet == writablePartitions && errors.Is(partition.Err, ErrLeaderNotAvailable) {
 			continue
 		}
 		ret = append(ret, partition.ID)
@@ -760,7 +759,7 @@ func (client *client) cachedLeader(topic string, partitionID int32) (*Broker, er
 	if partitions != nil {
 		metadata, ok := partitions[partitionID]
 		if ok {
-			if metadata.Err == ErrLeaderNotAvailable {
+			if errors.Is(metadata.Err, ErrLeaderNotAvailable) {
 				return nil, ErrLeaderNotAvailable
 			}
 			b := client.brokers[metadata.Leader]
@@ -798,7 +797,7 @@ func (client *client) getOffset(topic string, partitionID int32, time int64) (in
 		_ = broker.Close()
 		return -1, ErrIncompleteResponse
 	}
-	if block.Err != ErrNoError {
+	if !errors.Is(block.Err, ErrNoError) {
 		return -1, block.Err
 	}
 	if len(block.Offsets) != 1 {
@@ -877,6 +876,7 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 	}
 
 	broker := client.any()
+	brokerErrors := make([]error, 0)
 	for ; broker != nil && !pastDeadline(0); broker = client.any() {
 		allowAutoTopicCreation := client.conf.Metadata.AllowAutoTopicCreation
 		if len(topics) > 0 {
@@ -893,8 +893,9 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 			req.Version = 1
 		}
 		response, err := broker.GetMetadata(req)
-		switch err := err.(type) {
-		case nil:
+		var kerror KError
+		var packetEncodingError PacketEncodingError
+		if err == nil {
 			allKnownMetaData := len(topics) == 0
 			// valid response, use it
 			shouldRetry, err := client.updateMetadata(response, allKnownMetaData)
@@ -903,19 +904,17 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 				return retry(err) // note: err can be nil
 			}
 			return err
-
-		case PacketEncodingError:
+		} else if errors.As(err, &packetEncodingError) {
 			// didn't even send, return the error
 			return err
-
-		case KError:
+		} else if errors.As(err, &kerror) {
 			// if SASL auth error return as this _should_ be a non retryable err for all brokers
-			if err == ErrSASLAuthenticationFailed {
+			if errors.Is(err, ErrSASLAuthenticationFailed) {
 				Logger.Println("client/metadata failed SASL authentication")
 				return err
 			}
 
-			if err == ErrTopicAuthorizationFailed {
+			if errors.Is(err, ErrTopicAuthorizationFailed) {
 				Logger.Println("client is not authorized to access this topic. The topics were: ", topics)
 				return err
 			}
@@ -923,23 +922,24 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 			Logger.Printf("client/metadata got error from broker %d while fetching metadata: %v\n", broker.ID(), err)
 			_ = broker.Close()
 			client.deregisterBroker(broker)
-
-		default:
+		} else {
 			// some other error, remove that broker and try again
 			Logger.Printf("client/metadata got error from broker %d while fetching metadata: %v\n", broker.ID(), err)
+			brokerErrors = append(brokerErrors, err)
 			_ = broker.Close()
 			client.deregisterBroker(broker)
 		}
 	}
 
+	error := Wrap(ErrOutOfBrokers, brokerErrors...)
 	if broker != nil {
 		Logger.Printf("client/metadata not fetching metadata from broker %s as we would go past the metadata timeout\n", broker.addr)
-		return retry(ErrOutOfBrokers)
+		return retry(error)
 	}
 
 	Logger.Println("client/metadata no available broker to send metadata request to")
 	client.resurrectDeadBrokers()
-	return retry(ErrOutOfBrokers)
+	return retry(error)
 }
 
 // if no fatal error, returns a list of topics that need retrying due to ErrLeaderNotAvailable
@@ -996,7 +996,7 @@ func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bo
 		client.metadata[topic.Name] = make(map[int32]*PartitionMetadata, len(topic.Partitions))
 		for _, partition := range topic.Partitions {
 			client.metadata[topic.Name][partition.ID] = partition
-			if partition.Err == ErrLeaderNotAvailable {
+			if errors.Is(partition.Err, ErrLeaderNotAvailable) {
 				retry = true
 			}
 		}
@@ -1046,6 +1046,7 @@ func (client *client) getConsumerMetadata(consumerGroup string, attemptsRemainin
 		return nil, err
 	}
 
+	brokerErrors := make([]error, 0)
 	for broker := client.any(); broker != nil; broker = client.any() {
 		DebugLogger.Printf("client/coordinator requesting coordinator for consumergroup %s from %s\n", consumerGroup, broker.Addr())
 
@@ -1057,22 +1058,21 @@ func (client *client) getConsumerMetadata(consumerGroup string, attemptsRemainin
 		if err != nil {
 			Logger.Printf("client/coordinator request to broker %s failed: %s\n", broker.Addr(), err)
 
-			switch err.(type) {
-			case PacketEncodingError:
+			var packetEncodingError PacketEncodingError
+			if errors.As(err, &packetEncodingError) {
 				return nil, err
-			default:
+			} else {
 				_ = broker.Close()
+				brokerErrors = append(brokerErrors, err)
 				client.deregisterBroker(broker)
 				continue
 			}
 		}
 
-		switch response.Err {
-		case ErrNoError:
+		if errors.Is(response.Err, ErrNoError) {
 			DebugLogger.Printf("client/coordinator coordinator for consumergroup %s is #%d (%s)\n", consumerGroup, response.Coordinator.ID(), response.Coordinator.Addr())
 			return response, nil
-
-		case ErrConsumerCoordinatorNotAvailable:
+		} else if errors.Is(response.Err, ErrConsumerCoordinatorNotAvailable) {
 			Logger.Printf("client/coordinator coordinator for consumer group %s is not available\n", consumerGroup)
 
 			// This is very ugly, but this scenario will only happen once per cluster.
@@ -1084,18 +1084,17 @@ func (client *client) getConsumerMetadata(consumerGroup string, attemptsRemainin
 			}
 
 			return retry(ErrConsumerCoordinatorNotAvailable)
-		case ErrGroupAuthorizationFailed:
+		} else if errors.Is(response.Err, ErrGroupAuthorizationFailed) {
 			Logger.Printf("client was not authorized to access group %s while attempting to find coordinator", consumerGroup)
 			return retry(ErrGroupAuthorizationFailed)
-
-		default:
+		} else {
 			return nil, response.Err
 		}
 	}
 
 	Logger.Println("client/coordinator no available broker to send consumer metadata request to")
 	client.resurrectDeadBrokers()
-	return retry(ErrOutOfBrokers)
+	return retry(Wrap(ErrOutOfBrokers, brokerErrors...))
 }
 
 // nopCloserClient embeds an existing Client, but disables
diff --git a/consumer.go b/consumer.go
index df13c29..dd1d365 100644
--- a/consumer.go
+++ b/consumer.go
@@ -697,7 +697,7 @@ func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*Consu
 		return nil, ErrIncompleteResponse
 	}
 
-	if block.Err != ErrNoError {
+	if !errors.Is(block.Err, ErrNoError) {
 		return nil, block.Err
 	}
 
@@ -983,25 +983,24 @@ func (bc *brokerConsumer) handleResponses() {
 		// Discard any replica preference.
 		child.preferredReadReplica = invalidPreferredReplicaID
 
-		switch result {
-		case errTimedOut:
+		if errors.Is(result, errTimedOut) {
 			Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because consuming was taking too long\n",
 				bc.broker.ID(), child.topic, child.partition)
 			delete(bc.subscriptions, child)
-		case ErrOffsetOutOfRange:
+		} else if errors.Is(result, ErrOffsetOutOfRange) {
 			// there's no point in retrying this it will just fail the same way again
 			// shut it down and force the user to choose what to do
 			child.sendError(result)
 			Logger.Printf("consumer/%s/%d shutting down because %s\n", child.topic, child.partition, result)
 			close(child.trigger)
 			delete(bc.subscriptions, child)
-		case ErrUnknownTopicOrPartition, ErrNotLeaderForPartition, ErrLeaderNotAvailable, ErrReplicaNotAvailable:
+		} else if errors.Is(result, ErrUnknownTopicOrPartition) || errors.Is(result, ErrNotLeaderForPartition) || errors.Is(result, ErrLeaderNotAvailable) || errors.Is(result, ErrReplicaNotAvailable) {
 			// not an error, but does need redispatching
 			Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because %s\n",
 				bc.broker.ID(), child.topic, child.partition, result)
 			child.trigger <- none{}
 			delete(bc.subscriptions, child)
-		default:
+		} else {
 			// dunno, tell the user and try redispatching
 			child.sendError(result)
 			Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because %s\n",
diff --git a/consumer_group.go b/consumer_group.go
index 2355dbb..18f5878 100644
--- a/consumer_group.go
+++ b/consumer_group.go
@@ -190,7 +190,7 @@ func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler Co
 
 	// Init session
 	sess, err := c.newSession(ctx, topics, handler, c.config.Consumer.Group.Rebalance.Retry.Max)
-	if err == ErrClosedClient {
+	if errors.Is(err, ErrClosedClient) {
 		return ErrClosedConsumerGroup
 	} else if err != nil {
 		return err
@@ -282,7 +282,7 @@ func (c *consumerGroup) newSession(ctx context.Context, topics []string, handler
 		}
 		return nil, err
 	}
-	if join.Err != ErrNoError {
+	if !errors.Is(join.Err, ErrNoError) {
 		if consumerGroupJoinFailed != nil {
 			consumerGroupJoinFailed.Inc(1)
 		}
@@ -335,7 +335,7 @@ func (c *consumerGroup) newSession(ctx context.Context, topics []string, handler
 		}
 		return nil, err
 	}
-	if groupRequest.Err != ErrNoError {
+	if !errors.Is(groupRequest.Err, ErrNoError) {
 		if consumerGroupSyncFailed != nil {
 			consumerGroupSyncFailed.Inc(1)
 		}
@@ -498,7 +498,8 @@ func (c *consumerGroup) leave() error {
 }
 
 func (c *consumerGroup) handleError(err error, topic string, partition int32) {
-	if _, ok := err.(*ConsumerError); !ok && topic != "" && partition > -1 {
+	var consumerError *ConsumerError
+	if ok := errors.As(err, &consumerError); !ok && topic != "" && partition > -1 {
 		err = &ConsumerError{
 			Topic:     topic,
 			Partition: partition,
@@ -939,7 +940,7 @@ type consumerGroupClaim struct {
 
 func newConsumerGroupClaim(sess *consumerGroupSession, topic string, partition int32, offset int64) (*consumerGroupClaim, error) {
 	pcm, err := sess.parent.consumer.ConsumePartition(topic, partition, offset)
-	if err == ErrOffsetOutOfRange {
+	if errors.Is(err, ErrOffsetOutOfRange) {
 		offset = sess.parent.config.Consumer.Offsets.Initial
 		pcm, err = sess.parent.consumer.ConsumePartition(topic, partition, offset)
 	}
diff --git a/consumer_group_members.go b/consumer_group_members.go
index f4100f5..3b8ca36 100644
--- a/consumer_group_members.go
+++ b/consumer_group_members.go
@@ -1,5 +1,7 @@
 package sarama
 
+import "errors"
+
 // ConsumerGroupMemberMetadata holds the metadata for consumer group
 // https://github.com/apache/kafka/blob/trunk/clients/src/main/resources/common/message/ConsumerProtocolSubscription.json
 type ConsumerGroupMemberMetadata struct {
@@ -41,7 +43,7 @@ func (m *ConsumerGroupMemberMetadata) decode(pd packetDecoder) (err error) {
 			// permit missing data here in case of misbehaving 3rd party
 			// clients who incorrectly marked the member metadata as V1 in
 			// their JoinGroup request
-			if err == ErrInsufficientData {
+			if errors.Is(err, ErrInsufficientData) {
 				return nil
 			}
 			return err
diff --git a/create_partitions_response.go b/create_partitions_response.go
index 12ce788..235787f 100644
--- a/create_partitions_response.go
+++ b/create_partitions_response.go
@@ -84,6 +84,10 @@ func (t *TopicPartitionError) Error() string {
 	return text
 }
 
+func (t *TopicPartitionError) Unwrap() error {
+	return t.Err
+}
+
 func (t *TopicPartitionError) encode(pe packetEncoder) error {
 	pe.putInt16(int16(t.Err))
 
diff --git a/create_topics_response.go b/create_topics_response.go
index 7e1448a..6b940bf 100644
--- a/create_topics_response.go
+++ b/create_topics_response.go
@@ -98,6 +98,10 @@ func (t *TopicError) Error() string {
 	return text
 }
 
+func (t *TopicError) Unwrap() error {
+	return t.Err
+}
+
 func (t *TopicError) encode(pe packetEncoder, version int16) error {
 	pe.putInt16(int16(t.Err))
 
diff --git a/errors.go b/errors.go
index 0fca0a3..bf5d76b 100644
--- a/errors.go
+++ b/errors.go
@@ -3,6 +3,8 @@ package sarama
 import (
 	"errors"
 	"fmt"
+
+	"github.com/hashicorp/go-multierror"
 )
 
 // ErrOutOfBrokers is the error returned when the client has run out of brokers to talk to because all of them errored
@@ -55,6 +57,48 @@ var ErrNoTopicsToUpdateMetadata = errors.New("kafka: no specific topics to updat
 // ErrUnknownScramMechanism is returned when user tries to AlterUserScramCredentials with unknown SCRAM mechanism
 var ErrUnknownScramMechanism = errors.New("kafka: unknown SCRAM mechanism provided")
 
+// ErrReassignPartitions is returned when altering partition assignments for a topic fails
+var ErrReassignPartitions = errors.New("failed to reassign partitions for topic")
+
+// ErrDeleteRecords is the type of error returned when fail to delete the required records
+var ErrDeleteRecords = errors.New("kafka server: failed to delete records")
+
+// The formatter used to format multierrors
+var MultiErrorFormat multierror.ErrorFormatFunc
+
+type sentinelError struct {
+	sentinel error
+	wrapped  error
+}
+
+func (err sentinelError) Error() string {
+	if err.wrapped != nil {
+		return fmt.Sprintf("%s: %v", err.sentinel, err.wrapped)
+	} else {
+		return fmt.Sprintf("%s", err.sentinel)
+	}
+}
+
+func (err sentinelError) Is(target error) bool {
+	return errors.Is(err.sentinel, target) || errors.Is(err.wrapped, target)
+}
+
+func (err sentinelError) Unwrap() error {
+	return err.wrapped
+}
+
+func Wrap(sentinel error, wrapped ...error) sentinelError {
+	return sentinelError{sentinel: sentinel, wrapped: multiError(wrapped...)}
+}
+
+func multiError(wrapped ...error) error {
+	merr := multierror.Append(nil, wrapped...)
+	if MultiErrorFormat != nil {
+		merr.ErrorFormat = MultiErrorFormat
+	}
+	return merr.ErrorOrNil()
+}
+
 // PacketEncodingError is returned from a failure while encoding a Kafka packet. This can happen, for example,
 // if you try to encode a string over 2^15 characters in length, since Kafka's encoding rules do not permit that.
 type PacketEncodingError struct {
@@ -87,44 +131,6 @@ func (err ConfigurationError) Error() string {
 // See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes
 type KError int16
 
-// MultiError is used to contain multi error.
-type MultiError struct {
-	Errors *[]error
-}
-
-func (mErr MultiError) Error() string {
-	errString := ""
-	for _, err := range *mErr.Errors {
-		errString += err.Error() + ","
-	}
-	return errString
-}
-
-func (mErr MultiError) PrettyError() string {
-	errString := ""
-	for _, err := range *mErr.Errors {
-		errString += err.Error() + "\n"
-	}
-	return errString
-}
-
-// ErrDeleteRecords is the type of error returned when fail to delete the required records
-type ErrDeleteRecords struct {
-	MultiError
-}
-
-func (err ErrDeleteRecords) Error() string {
-	return "kafka server: failed to delete records " + err.MultiError.Error()
-}
-
-type ErrReassignPartitions struct {
-	MultiError
-}
-
-func (err ErrReassignPartitions) Error() string {
-	return fmt.Sprintf("failed to reassign partitions for topic: \n%s", err.MultiError.PrettyError())
-}
-
 // Numeric error codes returned by the Kafka server.
 const (
 	ErrNoError                            KError = 0
diff --git a/fetch_response.go b/fetch_response.go
index 1b02b00..5a392ec 100644
--- a/fetch_response.go
+++ b/fetch_response.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"sort"
 	"time"
 )
@@ -112,7 +113,7 @@ func (b *FetchResponseBlock) decode(pd packetDecoder, version int16) (err error)
 		records := &Records{}
 		if err := records.decode(recordsDecoder); err != nil {
 			// If we have at least one decoded records, this is not an error
-			if err == ErrInsufficientData {
+			if errors.Is(err, ErrInsufficientData) {
 				if len(b.RecordsSet) == 0 {
 					b.Partial = true
 				}
diff --git a/go.mod b/go.mod
index b949f94..8985389 100644
--- a/go.mod
+++ b/go.mod
@@ -11,6 +11,7 @@ require (
 	github.com/fortytw2/leaktest v1.3.0
 	github.com/frankban/quicktest v1.14.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/jcmturner/gofork v1.0.0
 	github.com/jcmturner/gokrb5/v8 v8.4.2
 	github.com/klauspost/compress v1.14.4
diff --git a/go.sum b/go.sum
index 6606a68..0ef5b38 100644
--- a/go.sum
+++ b/go.sum
@@ -23,6 +23,10 @@ github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 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/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
diff --git a/message_set.go b/message_set.go
index 6523ec2..5bc2112 100644
--- a/message_set.go
+++ b/message_set.go
@@ -1,5 +1,7 @@
 package sarama
 
+import "errors"
+
 type MessageBlock struct {
 	Offset int64
 	Msg    *Message
@@ -70,7 +72,7 @@ func (ms *MessageSet) decode(pd packetDecoder) (err error) {
 	for pd.remaining() > 0 {
 		magic, err := magicValue(pd)
 		if err != nil {
-			if err == ErrInsufficientData {
+			if errors.Is(err, ErrInsufficientData) {
 				ms.PartialTrailingMessage = true
 				return nil
 			}
@@ -83,10 +85,9 @@ func (ms *MessageSet) decode(pd packetDecoder) (err error) {
 
 		msb := new(MessageBlock)
 		err = msb.decode(pd)
-		switch err {
-		case nil:
+		if err == nil {
 			ms.Messages = append(ms.Messages, msb)
-		case ErrInsufficientData:
+		} else if errors.Is(err, ErrInsufficientData) {
 			// As an optimization the server is allowed to return a partial message at the
 			// end of the message set. Clients should handle this case. So we just ignore such things.
 			if msb.Offset == -1 {
@@ -96,7 +97,7 @@ func (ms *MessageSet) decode(pd packetDecoder) (err error) {
 				ms.PartialTrailingMessage = true
 			}
 			return nil
-		default:
+		} else {
 			return err
 		}
 	}
diff --git a/mockbroker.go b/mockbroker.go
index 9166f6e..628c3cb 100644
--- a/mockbroker.go
+++ b/mockbroker.go
@@ -3,6 +3,7 @@ package sarama
 import (
 	"bytes"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -359,9 +360,10 @@ func (b *MockBroker) defaultRequestHandler(req *request) (res encoderWithHeader)
 
 func (b *MockBroker) serverError(err error) {
 	isConnectionClosedError := false
-	if _, ok := err.(*net.OpError); ok {
+	opError := &net.OpError{}
+	if errors.As(err, &opError) {
 		isConnectionClosedError = true
-	} else if err == io.EOF {
+	} else if errors.Is(err, io.EOF) {
 		isConnectionClosedError = true
 	} else if err.Error() == "use of closed network connection" {
 		isConnectionClosedError = true
diff --git a/mocks/async_producer.go b/mocks/async_producer.go
index 418f077..a139fb6 100644
--- a/mocks/async_producer.go
+++ b/mocks/async_producer.go
@@ -1,6 +1,7 @@
 package mocks
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/Shopify/sarama"
@@ -78,7 +79,7 @@ func NewAsyncProducer(t ErrorReporter, config *sarama.Config) *AsyncProducer {
 							mp.errors <- &sarama.ProducerError{Err: err, Msg: msg}
 						}
 					}
-					if expectation.Result == errProduceSuccess {
+					if errors.Is(expectation.Result, errProduceSuccess) {
 						mp.lastOffset++
 						if config.Producer.Return.Successes {
 							msg.Offset = mp.lastOffset
diff --git a/mocks/mocks.go b/mocks/mocks.go
index 25671bc..c9872a7 100644
--- a/mocks/mocks.go
+++ b/mocks/mocks.go
@@ -44,7 +44,7 @@ func messageValueChecker(f ValueChecker) MessageChecker {
 	return func(msg *sarama.ProducerMessage) error {
 		val, err := msg.Value.Encode()
 		if err != nil {
-			return fmt.Errorf("Input message encoding failed: %s", err.Error())
+			return fmt.Errorf("Input message encoding failed: %w", err)
 		}
 		return f(val)
 	}
diff --git a/mocks/sync_producer.go b/mocks/sync_producer.go
index 67e5352..ef7d679 100644
--- a/mocks/sync_producer.go
+++ b/mocks/sync_producer.go
@@ -1,6 +1,7 @@
 package mocks
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/Shopify/sarama"
@@ -68,7 +69,7 @@ func (sp *SyncProducer) SendMessage(msg *sarama.ProducerMessage) (partition int3
 				return -1, -1, errCheck
 			}
 		}
-		if expectation.Result == errProduceSuccess {
+		if errors.Is(expectation.Result, errProduceSuccess) {
 			sp.lastOffset++
 			msg.Offset = sp.lastOffset
 			return 0, msg.Offset, nil
@@ -106,7 +107,7 @@ func (sp *SyncProducer) SendMessages(msgs []*sarama.ProducerMessage) error {
 					return errCheck
 				}
 			}
-			if expectation.Result != errProduceSuccess {
+			if !errors.Is(expectation.Result, errProduceSuccess) {
 				return expectation.Result
 			}
 			sp.lastOffset++
diff --git a/record_batch.go b/record_batch.go
index c653763..d7ef261 100644
--- a/record_batch.go
+++ b/record_batch.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"fmt"
 	"time"
 )
@@ -167,7 +168,7 @@ func (b *RecordBatch) decode(pd packetDecoder) (err error) {
 	bufSize := int(batchLen) - recordBatchOverhead
 	recBuffer, err := pd.getRawBytes(bufSize)
 	if err != nil {
-		if err == ErrInsufficientData {
+		if errors.Is(err, ErrInsufficientData) {
 			b.PartialTrailingRecord = true
 			b.Records = nil
 			return nil
@@ -186,7 +187,7 @@ func (b *RecordBatch) decode(pd packetDecoder) (err error) {
 
 	b.recordsLen = len(recBuffer)
 	err = decode(recBuffer, recordsArray(b.Records))
-	if err == ErrInsufficientData {
+	if errors.Is(err, ErrInsufficientData) {
 		b.PartialTrailingRecord = true
 		b.Records = nil
 		return nil


Test patch:
diff --git a/admin_test.go b/admin_test.go
index 9441142..683e220 100644
--- a/admin_test.go
+++ b/admin_test.go
@@ -47,7 +47,7 @@ func TestClusterAdminInvalidController(t *testing.T) {
 		t.Fatal(errors.New("controller not set still cluster admin was created"))
 	}
 
-	if err != ErrControllerNotAvailable {
+	if !errors.Is(err, ErrControllerNotAvailable) {
 		t.Fatal(err)
 	}
 }
@@ -243,7 +243,7 @@ func TestClusterAdminDeleteEmptyTopic(t *testing.T) {
 	}
 
 	err = admin.DeleteTopic("")
-	if err != ErrInvalidTopic {
+	if !errors.Is(err, ErrInvalidTopic) {
 		t.Fatal(err)
 	}
 
@@ -303,7 +303,7 @@ func TestClusterAdminCreatePartitionsWithDiffVersion(t *testing.T) {
 	}
 
 	err = admin.CreatePartitions("my_topic", 3, nil, false)
-	if err != ErrUnsupportedVersion {
+	if !errors.Is(err, ErrUnsupportedVersion) {
 		t.Fatal(err)
 	}
 
@@ -639,19 +639,20 @@ func TestClusterAdminDeleteRecordsWithDiffVersion(t *testing.T) {
 	partitionOffset[3] = 1000
 
 	err = admin.DeleteRecords(topicName, partitionOffset)
+	if err == nil {
+		t.Fatal("expected an ErrDeleteRecords")
+	}
+
 	if !strings.HasPrefix(err.Error(), "kafka server: failed to delete records") {
 		t.Fatal(err)
 	}
-	deleteRecordsError, ok := err.(ErrDeleteRecords)
 
-	if !ok {
+	if !errors.Is(err, ErrDeleteRecords) {
 		t.Fatal(err)
 	}
 
-	for _, err := range *deleteRecordsError.Errors {
-		if err != ErrUnsupportedVersion {
-			t.Fatal(err)
-		}
+	if !errors.Is(err, ErrUnsupportedVersion) {
+		t.Fatal(err)
 	}
 
 	err = admin.Close()
@@ -1547,7 +1548,7 @@ func TestDeleteOffset(t *testing.T) {
 	handlerMap["DeleteOffsetsRequest"] = NewMockDeleteOffsetRequest(t).SetDeletedOffset(ErrNotCoordinatorForConsumer, topic, partition, ErrNoError)
 	seedBroker.SetHandlerByMap(handlerMap)
 	err = admin.DeleteConsumerGroupOffset(group, topic, partition)
-	if err != ErrNotCoordinatorForConsumer {
+	if !errors.Is(err, ErrNotCoordinatorForConsumer) {
 		t.Fatalf("DeleteConsumerGroupOffset should have failed with error %v", ErrNotCoordinatorForConsumer)
 	}
 
@@ -1555,7 +1556,7 @@ func TestDeleteOffset(t *testing.T) {
 	handlerMap["DeleteOffsetsRequest"] = NewMockDeleteOffsetRequest(t).SetDeletedOffset(ErrNoError, topic, partition, ErrGroupSubscribedToTopic)
 	seedBroker.SetHandlerByMap(handlerMap)
 	err = admin.DeleteConsumerGroupOffset(group, topic, partition)
-	if err != ErrGroupSubscribedToTopic {
+	if !errors.Is(err, ErrGroupSubscribedToTopic) {
 		t.Fatalf("DeleteConsumerGroupOffset should have failed with error %v", ErrGroupSubscribedToTopic)
 	}
 }
@@ -1643,7 +1644,7 @@ func TestDescribeLogDirs(t *testing.T) {
 		t.Fatalf("Expected log dirs for broker %v to be returned, but it did not, got %v", seedBroker.BrokerID(), len(logDirs))
 	}
 	logDirsBroker := logDirs[0]
-	if logDirsBroker.ErrorCode != ErrNoError {
+	if !errors.Is(logDirsBroker.ErrorCode, ErrNoError) {
 		t.Fatalf("Expected no error for broker %v, but it was %v", seedBroker.BrokerID(), logDirsBroker.ErrorCode)
 	}
 	if logDirsBroker.Path != "/tmp/logs" {
diff --git a/async_producer_test.go b/async_producer_test.go
index 60810b3..1911c26 100644
--- a/async_producer_test.go
+++ b/async_producer_test.go
@@ -740,7 +740,7 @@ func TestAsyncProducerOutOfRetries(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		select {
 		case msg := <-producer.Errors():
-			if msg.Err != ErrNotLeaderForPartition {
+			if !errors.Is(msg.Err, ErrNotLeaderForPartition) {
 				t.Error(msg.Err)
 			}
 		case <-producer.Successes():
@@ -916,7 +916,7 @@ func TestAsyncProducerRetryShutdown(t *testing.T) {
 	time.Sleep(5 * time.Millisecond) // let the shutdown goroutine kick in
 
 	producer.Input() <- &ProducerMessage{Topic: "FOO"}
-	if err := <-producer.Errors(); err.Err != ErrShuttingDown {
+	if err := <-producer.Errors(); !errors.Is(err.Err, ErrShuttingDown) {
 		t.Error(err)
 	}
 
@@ -1457,6 +1457,14 @@ func TestAsyncProducerInterceptors(t *testing.T) {
 	}
 }
 
+func TestProducerError(t *testing.T) {
+	t.Parallel()
+	err := ProducerError{Err: ErrOutOfBrokers}
+	if !errors.Is(err, ErrOutOfBrokers) {
+		t.Error("unexpected errors.Is")
+	}
+}
+
 // This example shows how to use the producer while simultaneously
 // reading the Errors channel to know about any failures.
 func ExampleAsyncProducer_select() {
diff --git a/broker_test.go b/broker_test.go
index 909995b..378b927 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -188,7 +188,7 @@ func TestBrokerFailedRequest(t *testing.T) {
 			}
 			err = broker.Close()
 			if err != nil {
-				if tt.stopBroker && err == ErrNotConnected {
+				if tt.stopBroker && errors.Is(err, ErrNotConnected) {
 					// We expect the broker to not close properly
 					return
 				}
@@ -331,8 +331,8 @@ func TestSASLOAuthBearer(t *testing.T) {
 
 			err = broker.authenticateViaSASL()
 
-			if test.expectedBrokerError != ErrNoError {
-				if test.expectedBrokerError != err {
+			if !errors.Is(test.expectedBrokerError, ErrNoError) {
+				if !errors.Is(err, test.expectedBrokerError) {
 					t.Errorf("[%d]:[%s] Expected %s auth error, got %s\n", i, test.name, test.expectedBrokerError, err)
 				}
 			} else if test.expectClientErr && err == nil {
@@ -432,10 +432,10 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 			mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).SetAuthBytes([]byte(test.scramChallengeResp))
 			mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).SetEnabledMechanisms([]string{SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512})
 
-			if test.mockSASLAuthErr != ErrNoError {
+			if !errors.Is(test.mockSASLAuthErr, ErrNoError) {
 				mockSASLAuthResponse = mockSASLAuthResponse.SetError(test.mockSASLAuthErr)
 			}
-			if test.mockHandshakeErr != ErrNoError {
+			if !errors.Is(test.mockHandshakeErr, ErrNoError) {
 				mockSASLHandshakeResponse = mockSASLHandshakeResponse.SetError(test.mockHandshakeErr)
 			}
 
@@ -466,12 +466,12 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 
 			err = broker.authenticateViaSASL()
 
-			if test.mockSASLAuthErr != ErrNoError {
-				if test.mockSASLAuthErr != err {
+			if !errors.Is(test.mockSASLAuthErr, ErrNoError) {
+				if !errors.Is(err, test.mockSASLAuthErr) {
 					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 {
+			} else if !errors.Is(test.mockHandshakeErr, ErrNoError) {
+				if !errors.Is(err, test.mockHandshakeErr) {
 					t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.mockHandshakeErr, err)
 				}
 			} else if test.expectClientErr && err == nil {
@@ -527,14 +527,14 @@ func TestSASLPlainAuth(t *testing.T) {
 			mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).
 				SetAuthBytes([]byte(`response_payload`))
 
-			if test.mockAuthErr != ErrNoError {
+			if !errors.Is(test.mockAuthErr, ErrNoError) {
 				mockSASLAuthResponse = mockSASLAuthResponse.SetError(test.mockAuthErr)
 			}
 
 			mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).
 				SetEnabledMechanisms([]string{SASLTypePlaintext})
 
-			if test.mockHandshakeErr != ErrNoError {
+			if !errors.Is(test.mockHandshakeErr, ErrNoError) {
 				mockSASLHandshakeResponse = mockSASLHandshakeResponse.SetError(test.mockHandshakeErr)
 			}
 
@@ -595,12 +595,12 @@ func TestSASLPlainAuth(t *testing.T) {
 				}
 			}
 
-			if test.mockAuthErr != ErrNoError {
-				if test.mockAuthErr != err {
+			if !errors.Is(test.mockAuthErr, ErrNoError) {
+				if !errors.Is(err, test.mockAuthErr) {
 					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 {
+			} else if !errors.Is(test.mockHandshakeErr, ErrNoError) {
+				if !errors.Is(err, test.mockHandshakeErr) {
 					t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.mockHandshakeErr, err)
 				}
 			} else if test.expectClientErr && err == nil {
diff --git a/client_test.go b/client_test.go
index f5253a5..49d79d0 100644
--- a/client_test.go
+++ b/client_test.go
@@ -1,9 +1,11 @@
 package sarama
 
 import (
+	"errors"
 	"io"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"testing"
 	"time"
 )
@@ -95,7 +97,7 @@ func TestClientDoesntCachePartitionsForTopicsWithErrors(t *testing.T) {
 
 	partitions, err := client.Partitions("unknown")
 
-	if err != ErrUnknownTopicOrPartition {
+	if !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("Expected ErrUnknownTopicOrPartition, found", err)
 	}
 	if partitions != nil {
@@ -114,7 +116,7 @@ func TestClientDoesntCachePartitionsForTopicsWithErrors(t *testing.T) {
 
 	// Should not use cache for unknown topic
 	partitions, err = client.Partitions("unknown")
-	if err != ErrUnknownTopicOrPartition {
+	if !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("Expected ErrUnknownTopicOrPartition, found", err)
 	}
 	if partitions != nil {
@@ -381,7 +383,7 @@ func TestClientReceivingUnknownTopicWithBackoffFunc(t *testing.T) {
 	seedBroker.Returns(metadataUnknownTopic)
 	seedBroker.Returns(metadataUnknownTopic)
 
-	if err := client.RefreshMetadata("new_topic"); err != ErrUnknownTopicOrPartition {
+	if err := client.RefreshMetadata("new_topic"); !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("ErrUnknownTopicOrPartition expected, got", err)
 	}
 
@@ -414,7 +416,7 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
 	seedBroker.Returns(metadataUnknownTopic)
 	seedBroker.Returns(metadataUnknownTopic)
 
-	if err := client.RefreshMetadata("new_topic"); err != ErrUnknownTopicOrPartition {
+	if err := client.RefreshMetadata("new_topic"); !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("ErrUnknownTopicOrPartition expected, got", err)
 	}
 
@@ -423,7 +425,7 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
 	seedBroker.Returns(metadataUnknownTopic)
 	seedBroker.Returns(metadataUnknownTopic)
 
-	if _, err = client.Leader("new_topic", 1); err != ErrUnknownTopicOrPartition {
+	if _, err = client.Leader("new_topic", 1); !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("Expected ErrUnknownTopicOrPartition, got", err)
 	}
 
@@ -478,7 +480,7 @@ func TestClientReceivingPartialMetadata(t *testing.T) {
 
 	// Still no leader for the partition, so asking for it should return an error.
 	_, err = client.Leader("new_topic", 1)
-	if err != ErrLeaderNotAvailable {
+	if !errors.Is(err, ErrLeaderNotAvailable) {
 		t.Error("Expected ErrLeaderNotAvailable, got", err)
 	}
 
@@ -620,7 +622,7 @@ func TestClientGetBroker(t *testing.T) {
 		t.Error(err)
 	}
 	_, err = client.Broker(leader.BrokerID())
-	if err != ErrBrokerNotFound {
+	if !errors.Is(err, ErrBrokerNotFound) {
 		t.Errorf("Expected Broker(brokerID) to return %v found %v", ErrBrokerNotFound, err)
 	}
 }
@@ -726,7 +728,7 @@ func TestClientController(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer safeClose(t, client2)
-		if _, err = client2.Controller(); err != ErrUnsupportedVersion {
+		if _, err = client2.Controller(); !errors.Is(err, ErrUnsupportedVersion) {
 			t.Errorf("Expected Controller() to return %s, found %s", ErrUnsupportedVersion, err)
 		}
 	})
@@ -807,7 +809,7 @@ func TestClientMetadataTimeout(t *testing.T) {
 				if err == nil {
 					t.Fatal("Expected failed RefreshMetadata, got nil")
 				}
-				if err != ErrOutOfBrokers {
+				if !errors.Is(err, ErrOutOfBrokers) {
 					t.Error("Expected failed RefreshMetadata with ErrOutOfBrokers, got:", err)
 				}
 			case <-time.After(maxRefreshDuration):
@@ -996,3 +998,74 @@ func TestClientAutorefreshShutdownRace(t *testing.T) {
 	// give the update time to happen so we get a panic if it's still running (which it shouldn't)
 	time.Sleep(10 * time.Millisecond)
 }
+
+func TestClientConnectionRefused(t *testing.T) {
+	t.Parallel()
+	seedBroker := NewMockBroker(t, 1)
+	seedBroker.Close()
+
+	_, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
+	if !errors.Is(err, ErrOutOfBrokers) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !errors.Is(err, syscall.ECONNREFUSED) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestClientCoordinatorConnectionRefused(t *testing.T) {
+	t.Parallel()
+	seedBroker := NewMockBroker(t, 1)
+	seedBroker.Returns(new(MetadataResponse))
+
+	client, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	seedBroker.Close()
+
+	_, err = client.Coordinator("my_group")
+
+	if !errors.Is(err, ErrOutOfBrokers) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !errors.Is(err, syscall.ECONNREFUSED) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	safeClose(t, client)
+}
+
+func TestInitProducerIDConnectionRefused(t *testing.T) {
+	t.Parallel()
+	seedBroker := NewMockBroker(t, 1)
+	seedBroker.Returns(&MetadataResponse{Version: 1})
+
+	config := NewTestConfig()
+	config.Producer.Idempotent = true
+	config.Version = V0_11_0_0
+	config.Producer.RequiredAcks = WaitForAll
+	config.Net.MaxOpenRequests = 1
+
+	client, err := NewClient([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	seedBroker.Close()
+
+	_, err = client.InitProducerID()
+
+	if !errors.Is(err, ErrOutOfBrokers) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !errors.Is(err, io.EOF) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	safeClose(t, client)
+}
diff --git a/config_test.go b/config_test.go
index 391b884..995761e 100644
--- a/config_test.go
+++ b/config_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"os"
 	"testing"
 
@@ -22,7 +23,9 @@ func TestInvalidClientIDConfigValidates(t *testing.T) {
 	t.Parallel()
 	config := NewTestConfig()
 	config.ClientID = "foo:bar"
-	if err := config.Validate(); string(err.(ConfigurationError)) != "ClientID is invalid" {
+	err := config.Validate()
+	var target ConfigurationError
+	if !errors.As(err, &target) || string(target) != "ClientID is invalid" {
 		t.Error("Expected invalid ClientID, got ", err)
 	}
 }
@@ -31,7 +34,9 @@ func TestEmptyClientIDConfigValidates(t *testing.T) {
 	t.Parallel()
 	config := NewTestConfig()
 	config.ClientID = ""
-	if err := config.Validate(); string(err.(ConfigurationError)) != "ClientID is invalid" {
+	err := config.Validate()
+	var target ConfigurationError
+	if !errors.As(err, &target) || string(target) != "ClientID is invalid" {
 		t.Error("Expected invalid ClientID, got ", err)
 	}
 }
@@ -232,7 +237,9 @@ func TestNetConfigValidates(t *testing.T) {
 	for i, test := range tests {
 		c := NewTestConfig()
 		test.cfg(c)
-		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+		err := c.Validate()
+		var target ConfigurationError
+		if !errors.As(err, &target) || string(target) != test.err {
 			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
 		}
 	}
@@ -271,7 +278,9 @@ func TestMetadataConfigValidates(t *testing.T) {
 	for i, test := range tests {
 		c := NewTestConfig()
 		test.cfg(c)
-		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+		err := c.Validate()
+		var target ConfigurationError
+		if !errors.As(err, &target) || string(target) != test.err {
 			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
 		}
 	}
@@ -296,7 +305,9 @@ func TestAdminConfigValidates(t *testing.T) {
 	for i, test := range tests {
 		c := NewTestConfig()
 		test.cfg(c)
-		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+		err := c.Validate()
+		var target ConfigurationError
+		if !errors.As(err, &target) || string(target) != test.err {
 			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
 		}
 	}
@@ -426,7 +437,9 @@ func TestProducerConfigValidates(t *testing.T) {
 	for i, test := range tests {
 		c := NewTestConfig()
 		test.cfg(c)
-		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+		err := c.Validate()
+		var target ConfigurationError
+		if !errors.As(err, &target) || string(target) != test.err {
 			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
 		}
 	}
@@ -460,7 +473,9 @@ func TestConsumerConfigValidates(t *testing.T) {
 	for i, test := range tests {
 		c := NewTestConfig()
 		test.cfg(c)
-		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+		err := c.Validate()
+		var target ConfigurationError
+		if !errors.As(err, &target) || string(target) != test.err {
 			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
 		}
 	}
@@ -470,7 +485,9 @@ func TestLZ4ConfigValidation(t *testing.T) {
 	t.Parallel()
 	config := NewTestConfig()
 	config.Producer.Compression = CompressionLZ4
-	if err := config.Validate(); string(err.(ConfigurationError)) != "lz4 compression requires Version >= V0_10_0_0" {
+	err := config.Validate()
+	var target ConfigurationError
+	if !errors.As(err, &target) || string(target) != "lz4 compression requires Version >= V0_10_0_0" {
 		t.Error("Expected invalid lz4/kafka version error, got ", err)
 	}
 	config.Version = V0_10_0_0
@@ -483,7 +500,9 @@ func TestZstdConfigValidation(t *testing.T) {
 	t.Parallel()
 	config := NewTestConfig()
 	config.Producer.Compression = CompressionZSTD
-	if err := config.Validate(); string(err.(ConfigurationError)) != "zstd compression requires Version >= V2_1_0_0" {
+	err := config.Validate()
+	var target ConfigurationError
+	if !errors.As(err, &target) || string(target) != "zstd compression requires Version >= V2_1_0_0" {
 		t.Error("Expected invalid zstd/kafka version error, got ", err)
 	}
 	config.Version = V2_1_0_0
diff --git a/consumer_metadata_response_test.go b/consumer_metadata_response_test.go
index 6b0f782..7e41752 100644
--- a/consumer_metadata_response_test.go
+++ b/consumer_metadata_response_test.go
@@ -1,6 +1,9 @@
 package sarama
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 var (
 	consumerMetadataResponseError = []byte{
@@ -28,7 +31,7 @@ func TestConsumerMetadataResponseError(t *testing.T) {
 		t.Error("could not decode: ", err)
 	}
 
-	if decodedResp.Err != ErrOffsetsLoadInProgress {
+	if !errors.Is(decodedResp.Err, ErrOffsetsLoadInProgress) {
 		t.Errorf("got %s, want %s", decodedResp.Err, ErrOffsetsLoadInProgress)
 	}
 }
diff --git a/consumer_test.go b/consumer_test.go
index 038bbd5..ea015eb 100644
--- a/consumer_test.go
+++ b/consumer_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"log"
 	"os"
 	"os/signal"
@@ -330,7 +331,9 @@ func TestConsumerDuplicate(t *testing.T) {
 	pc2, err := c.ConsumePartition("my_topic", 0, 0)
 
 	// Then
-	if pc2 != nil || err != ConfigurationError("That topic/partition is already being consumed") {
+	var target ConfigurationError
+	ok := errors.As(err, &target)
+	if pc2 != nil || !ok || string(target) != "That topic/partition is already being consumed" {
 		t.Fatal("A partition cannot be consumed twice at the same time")
 	}
 
@@ -380,7 +383,7 @@ func runConsumerLeaderRefreshErrorTestWithConfig(t *testing.T, config *Config) {
 		"FetchRequest": NewMockWrapper(fetchResponse2),
 	})
 
-	if consErr := <-pc.Errors(); consErr.Err != ErrOutOfBrokers {
+	if consErr := <-pc.Errors(); !errors.Is(consErr.Err, ErrOutOfBrokers) {
 		t.Errorf("Unexpected error: %v", consErr.Err)
 	}
 
@@ -462,7 +465,7 @@ func TestConsumerInvalidTopic(t *testing.T) {
 	pc, err := c.ConsumePartition("my_topic", 0, OffsetOldest)
 
 	// Then
-	if pc != nil || err != ErrUnknownTopicOrPartition {
+	if pc != nil || !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Errorf("Should fail with, err=%v", err)
 	}
 
@@ -514,7 +517,7 @@ func TestConsumerClosePartitionWithoutLeader(t *testing.T) {
 	})
 
 	// When
-	if consErr := <-pc.Errors(); consErr.Err != ErrOutOfBrokers {
+	if consErr := <-pc.Errors(); !errors.Is(consErr.Err, ErrOutOfBrokers) {
 		t.Errorf("Unexpected error: %v", consErr.Err)
 	}
 
@@ -1606,13 +1609,13 @@ func TestConsumerOffsetOutOfRange(t *testing.T) {
 	}
 
 	// When/Then
-	if _, err := master.ConsumePartition("my_topic", 0, 0); err != ErrOffsetOutOfRange {
+	if _, err := master.ConsumePartition("my_topic", 0, 0); !errors.Is(err, ErrOffsetOutOfRange) {
 		t.Fatal("Should return ErrOffsetOutOfRange, got:", err)
 	}
-	if _, err := master.ConsumePartition("my_topic", 0, 3456); err != ErrOffsetOutOfRange {
+	if _, err := master.ConsumePartition("my_topic", 0, 3456); !errors.Is(err, ErrOffsetOutOfRange) {
 		t.Fatal("Should return ErrOffsetOutOfRange, got:", err)
 	}
-	if _, err := master.ConsumePartition("my_topic", 0, -3); err != ErrOffsetOutOfRange {
+	if _, err := master.ConsumePartition("my_topic", 0, -3); !errors.Is(err, ErrOffsetOutOfRange) {
 		t.Fatal("Should return ErrOffsetOutOfRange, got:", err)
 	}
 
@@ -2091,3 +2094,11 @@ func TestConsumerInterceptors(t *testing.T) {
 		})
 	}
 }
+
+func TestConsumerError(t *testing.T) {
+	t.Parallel()
+	err := ConsumerError{Err: ErrOutOfBrokers}
+	if !errors.Is(err, ErrOutOfBrokers) {
+		t.Error("unexpected errors.Is")
+	}
+}
diff --git a/create_partitions_response_test.go b/create_partitions_response_test.go
index 9f9ee64..082bba0 100644
--- a/create_partitions_response_test.go
+++ b/create_partitions_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"reflect"
 	"testing"
 	"time"
@@ -59,6 +60,10 @@ func TestTopicPartitionError(t *testing.T) {
 		Err: ErrTopicAuthorizationFailed,
 	}
 
+	if !errors.Is(err, ErrTopicAuthorizationFailed) {
+		t.Errorf("unexpected errors.Is")
+	}
+
 	got := err.Error()
 	want := ErrTopicAuthorizationFailed.Error()
 	if got != want {
diff --git a/create_topics_response_test.go b/create_topics_response_test.go
index 73d8037..ac0942d 100644
--- a/create_topics_response_test.go
+++ b/create_topics_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"testing"
 	"time"
 )
@@ -59,6 +60,10 @@ func TestTopicError(t *testing.T) {
 		Err: ErrTopicAuthorizationFailed,
 	}
 
+	if !errors.Is(err, ErrTopicAuthorizationFailed) {
+		t.Errorf("unexpected errors.Is")
+	}
+
 	got := err.Error()
 	want := ErrTopicAuthorizationFailed.Error()
 	if got != want {
diff --git a/delete_groups_response_test.go b/delete_groups_response_test.go
index e524d28..9a59bb3 100644
--- a/delete_groups_response_test.go
+++ b/delete_groups_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"testing"
 )
 
@@ -43,7 +44,7 @@ func TestDeleteGroupsResponse(t *testing.T) {
 	if response.ThrottleTime != 0 {
 		t.Error("Expected no violation")
 	}
-	if response.GroupErrorCodes["foo"] != ErrClusterAuthorizationFailed {
+	if !errors.Is(response.GroupErrorCodes["foo"], ErrClusterAuthorizationFailed) {
 		t.Error("Expected error ErrClusterAuthorizationFailed, found:", response.GroupErrorCodes["foo"])
 	}
 
@@ -52,7 +53,7 @@ func TestDeleteGroupsResponse(t *testing.T) {
 	if response.ThrottleTime != 0 {
 		t.Error("Expected no violation")
 	}
-	if response.GroupErrorCodes["foo"] != ErrNoError {
+	if !errors.Is(response.GroupErrorCodes["foo"], ErrNoError) {
 		t.Error("Expected error ErrClusterAuthorizationFailed, found:", response.GroupErrorCodes["foo"])
 	}
 }
diff --git a/describe_groups_response_test.go b/describe_groups_response_test.go
index b63b25e..65617d4 100644
--- a/describe_groups_response_test.go
+++ b/describe_groups_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"reflect"
 	"testing"
 )
@@ -51,7 +52,7 @@ func TestDescribeGroupsResponse(t *testing.T) {
 	}
 
 	group0 := response.Groups[0]
-	if group0.Err != ErrNoError {
+	if !errors.Is(group0.Err, ErrNoError) {
 		t.Error("Unxpected groups[0].Err, found", group0.Err)
 	}
 	if group0.GroupId != "foo" {
@@ -83,7 +84,7 @@ func TestDescribeGroupsResponse(t *testing.T) {
 	}
 
 	group1 := response.Groups[1]
-	if group1.Err != ErrGroupAuthorizationFailed {
+	if !errors.Is(group1.Err, ErrGroupAuthorizationFailed) {
 		t.Error("Unxpected groups[1].Err, found", group0.Err)
 	}
 	if len(group1.Members) != 0 {
diff --git a/errors_test.go b/errors_test.go
new file mode 100644
index 0000000..d10649b
--- /dev/null
+++ b/errors_test.go
@@ -0,0 +1,65 @@
+package sarama
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"testing"
+)
+
+func TestSentinelWithWrappedError(t *testing.T) {
+	t.Parallel()
+	myNetError := &net.OpError{Op: "mock", Err: errors.New("op error")}
+	error := Wrap(ErrOutOfBrokers, myNetError)
+
+	expected := fmt.Sprintf("%s: 1 error occurred:\n\t* %s\n\n", ErrOutOfBrokers, myNetError)
+	actual := error.Error()
+	if actual != expected {
+		t.Errorf("unexpected value '%s' vs '%v'", expected, actual)
+	}
+
+	if !errors.Is(error, ErrOutOfBrokers) {
+		t.Error("errors.Is unexpected result")
+	}
+
+	if !errors.Is(error, myNetError) {
+		t.Error("errors.Is unexpected result")
+	}
+
+	var opError *net.OpError
+	if !errors.As(error, &opError) {
+		t.Error("errors.As unexpected result")
+	} else if opError != myNetError {
+		t.Error("errors.As wrong value")
+	}
+
+	unwrapped := errors.Unwrap(error)
+	if errors.Is(unwrapped, ErrOutOfBrokers) || !errors.Is(unwrapped, myNetError) {
+		t.Errorf("unexpected unwrapped value %v vs %vs", error, unwrapped)
+	}
+}
+
+func TestSentinelWithMultipleWrappedErrors(t *testing.T) {
+	t.Parallel()
+	myNetError := &net.OpError{}
+	myAddrError := &net.AddrError{}
+
+	error := Wrap(ErrOutOfBrokers, myNetError, myAddrError)
+
+	if !errors.Is(error, ErrOutOfBrokers) {
+		t.Error("errors.Is unexpected result")
+	}
+
+	if !errors.Is(error, myNetError) {
+		t.Error("errors.Is unexpected result")
+	}
+
+	if !errors.Is(error, myAddrError) {
+		t.Error("errors.Is unexpected result")
+	}
+
+	unwrapped := errors.Unwrap(error)
+	if errors.Is(unwrapped, ErrOutOfBrokers) || !errors.Is(unwrapped, myNetError) || !errors.Is(unwrapped, myAddrError) {
+		t.Errorf("unwrapped value unexpected result")
+	}
+}
diff --git a/fetch_response_test.go b/fetch_response_test.go
index badbac8..fb7b7ee 100644
--- a/fetch_response_test.go
+++ b/fetch_response_test.go
@@ -2,6 +2,7 @@ package sarama
 
 import (
 	"bytes"
+	"errors"
 	"testing"
 )
 
@@ -228,7 +229,7 @@ func TestOneMessageFetchResponse(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrOffsetOutOfRange {
+	if !errors.Is(block.Err, ErrOffsetOutOfRange) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
@@ -285,7 +286,7 @@ func TestOverflowMessageFetchResponse(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrOffsetOutOfRange {
+	if !errors.Is(block.Err, ErrOffsetOutOfRange) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
@@ -346,7 +347,7 @@ func TestOneRecordFetchResponse(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrOffsetOutOfRange {
+	if !errors.Is(block.Err, ErrOffsetOutOfRange) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
@@ -396,7 +397,7 @@ func TestPartailFetchResponse(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrNoError {
+	if !errors.Is(block.Err, ErrNoError) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
@@ -439,7 +440,7 @@ func TestEmptyRecordsFetchResponse(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrNoError {
+	if !errors.Is(block.Err, ErrNoError) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
@@ -485,7 +486,7 @@ func TestOneMessageFetchResponseV4(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrOffsetOutOfRange {
+	if !errors.Is(block.Err, ErrOffsetOutOfRange) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
@@ -552,7 +553,7 @@ func TestPreferredReplicaFetchResponseV11(t *testing.T) {
 	if block == nil {
 		t.Fatal("GetBlock didn't return block.")
 	}
-	if block.Err != ErrOffsetOutOfRange {
+	if !errors.Is(block.Err, ErrOffsetOutOfRange) {
 		t.Error("Decoding didn't produce correct error code.")
 	}
 	if block.HighWaterMarkOffset != 0x10101010 {
diff --git a/functional_client_test.go b/functional_client_test.go
index 0cf8c35..75eedac 100644
--- a/functional_client_test.go
+++ b/functional_client_test.go
@@ -4,6 +4,7 @@
 package sarama
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -20,7 +21,7 @@ func TestFuncConnectionFailure(t *testing.T) {
 	config.Metadata.Retry.Max = 1
 
 	_, err := NewClient([]string{FunctionalTestEnv.KafkaBrokerAddrs[0]}, config)
-	if err != ErrOutOfBrokers {
+	if !errors.Is(err, ErrOutOfBrokers) {
 		t.Fatal("Expected returned error to be ErrOutOfBrokers, but was: ", err)
 	}
 }
@@ -37,15 +38,15 @@ func TestFuncClientMetadata(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := client.RefreshMetadata("unknown_topic"); err != ErrUnknownTopicOrPartition {
+	if err := client.RefreshMetadata("unknown_topic"); !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("Expected ErrUnknownTopicOrPartition, got", err)
 	}
 
-	if _, err := client.Leader("unknown_topic", 0); err != ErrUnknownTopicOrPartition {
+	if _, err := client.Leader("unknown_topic", 0); !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("Expected ErrUnknownTopicOrPartition, got", err)
 	}
 
-	if _, err := client.Replicas("invalid/topic", 0); err != ErrUnknownTopicOrPartition && err != ErrInvalidTopic {
+	if _, err := client.Replicas("invalid/topic", 0); !errors.Is(err, ErrUnknownTopicOrPartition) && !errors.Is(err, ErrInvalidTopic) {
 		t.Error("Expected ErrUnknownTopicOrPartition or ErrInvalidTopic, got", err)
 	}
 
diff --git a/functional_consumer_group_test.go b/functional_consumer_group_test.go
index 13ed2e9..e6ac4dc 100644
--- a/functional_consumer_group_test.go
+++ b/functional_consumer_group_test.go
@@ -5,6 +5,7 @@ package sarama
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"reflect"
@@ -529,7 +530,7 @@ func (m *testFuncConsumerGroupMember) loop(topics []string) {
 		// set state to pre-consume
 		atomic.StoreInt32(&m.state, 1)
 
-		if err := m.Consume(ctx, topics, m); err == ErrClosedConsumerGroup {
+		if err := m.Consume(ctx, topics, m); errors.Is(err, ErrClosedConsumerGroup) {
 			return
 		} else if err != nil {
 			m.mu.Lock()
diff --git a/functional_consumer_test.go b/functional_consumer_test.go
index 442f2bb..f8de2b4 100644
--- a/functional_consumer_test.go
+++ b/functional_consumer_test.go
@@ -4,6 +4,7 @@
 package sarama
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"os"
@@ -25,11 +26,11 @@ func TestFuncConsumerOffsetOutOfRange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if _, err := consumer.ConsumePartition("test.1", 0, -10); err != ErrOffsetOutOfRange {
+	if _, err := consumer.ConsumePartition("test.1", 0, -10); !errors.Is(err, ErrOffsetOutOfRange) {
 		t.Error("Expected ErrOffsetOutOfRange, got:", err)
 	}
 
-	if _, err := consumer.ConsumePartition("test.1", 0, math.MaxInt64); err != ErrOffsetOutOfRange {
+	if _, err := consumer.ConsumePartition("test.1", 0, math.MaxInt64); !errors.Is(err, ErrOffsetOutOfRange) {
 		t.Error("Expected ErrOffsetOutOfRange, got:", err)
 	}
 
diff --git a/functional_producer_test.go b/functional_producer_test.go
index 4c44190..7c0cd28 100644
--- a/functional_producer_test.go
+++ b/functional_producer_test.go
@@ -4,6 +4,7 @@
 package sarama
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"strconv"
@@ -97,7 +98,7 @@ func TestFuncProducingToInvalidTopic(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if _, _, err := producer.SendMessage(&ProducerMessage{Topic: "in/valid"}); err != ErrUnknownTopicOrPartition && err != ErrInvalidTopic {
+	if _, _, err := producer.SendMessage(&ProducerMessage{Topic: "in/valid"}); !errors.Is(err, ErrUnknownTopicOrPartition) && !errors.Is(err, ErrInvalidTopic) {
 		t.Error("Expected ErrUnknownTopicOrPartition, found", err)
 	}
 
diff --git a/functional_test.go b/functional_test.go
index 509246c..5fbf5b9 100644
--- a/functional_test.go
+++ b/functional_test.go
@@ -5,6 +5,7 @@ package sarama
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"net/url"
@@ -377,11 +378,11 @@ func prepareTestTopics(ctx context.Context, env *testEnvironment) error {
 }
 
 func isTopicNotExistsErrorOrOk(err KError) bool {
-	return err == ErrUnknownTopicOrPartition || err == ErrInvalidTopic || err == ErrNoError
+	return errors.Is(err, ErrUnknownTopicOrPartition) || errors.Is(err, ErrInvalidTopic) || errors.Is(err, ErrNoError)
 }
 
 func isTopicExistsErrorOrOk(err KError) bool {
-	return err == ErrTopicAlreadyExists || err == ErrNoError
+	return errors.Is(err, ErrTopicAlreadyExists) || errors.Is(err, ErrNoError)
 }
 
 func checkKafkaVersion(t testing.TB, requiredVersion string) {
diff --git a/heartbeat_response_test.go b/heartbeat_response_test.go
index 6ca6eab..c23a7d5 100644
--- a/heartbeat_response_test.go
+++ b/heartbeat_response_test.go
@@ -1,6 +1,9 @@
 package sarama
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 var heartbeatResponseNoError = []byte{
 	0x00, 0x00,
@@ -10,7 +13,7 @@ func TestHeartbeatResponse(t *testing.T) {
 	t.Parallel()
 	response := new(HeartbeatResponse)
 	testVersionDecodable(t, "no error", response, heartbeatResponseNoError, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding error failed: no error expected but found", response.Err)
 	}
 }
diff --git a/join_group_response_test.go b/join_group_response_test.go
index 6ec1b64..72114f4 100644
--- a/join_group_response_test.go
+++ b/join_group_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"reflect"
 	"testing"
 )
@@ -61,7 +62,7 @@ func TestJoinGroupResponseV0(t *testing.T) {
 
 	response = new(JoinGroupResponse)
 	testVersionDecodable(t, "no error", response, joinGroupResponseV0_NoError, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding Err failed: no error expected but found", response.Err)
 	}
 	if response.GenerationId != 66051 {
@@ -79,7 +80,7 @@ func TestJoinGroupResponseV0(t *testing.T) {
 
 	response = new(JoinGroupResponse)
 	testVersionDecodable(t, "with error", response, joinGroupResponseV0_WithError, 0)
-	if response.Err != ErrInconsistentGroupProtocol {
+	if !errors.Is(response.Err, ErrInconsistentGroupProtocol) {
 		t.Error("Decoding Err failed: ErrInconsistentGroupProtocol expected but found", response.Err)
 	}
 	if response.GenerationId != 0 {
@@ -97,7 +98,7 @@ func TestJoinGroupResponseV0(t *testing.T) {
 
 	response = new(JoinGroupResponse)
 	testVersionDecodable(t, "with error", response, joinGroupResponseV0_Leader, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding Err failed: ErrNoError expected but found", response.Err)
 	}
 	if response.GenerationId != 66051 {
@@ -121,7 +122,7 @@ func TestJoinGroupResponseV1(t *testing.T) {
 	t.Parallel()
 	response := new(JoinGroupResponse)
 	testVersionDecodable(t, "no error", response, joinGroupResponseV1, 1)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding Err failed: no error expected but found", response.Err)
 	}
 	if response.GenerationId != 66051 {
@@ -151,7 +152,7 @@ func TestJoinGroupResponseV2(t *testing.T) {
 	if response.ThrottleTime != 100 {
 		t.Error("Decoding ThrottleTime failed, found:", response.ThrottleTime)
 	}
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding Err failed: no error expected but found", response.Err)
 	}
 	if response.GenerationId != 66051 {
diff --git a/leave_group_response_test.go b/leave_group_response_test.go
index 4a24334..1588742 100644
--- a/leave_group_response_test.go
+++ b/leave_group_response_test.go
@@ -1,6 +1,9 @@
 package sarama
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 var (
 	leaveGroupResponseNoError   = []byte{0x00, 0x00}
@@ -13,13 +16,13 @@ func TestLeaveGroupResponse(t *testing.T) {
 
 	response = new(LeaveGroupResponse)
 	testVersionDecodable(t, "no error", response, leaveGroupResponseNoError, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding error failed: no error expected but found", response.Err)
 	}
 
 	response = new(LeaveGroupResponse)
 	testVersionDecodable(t, "with error", response, leaveGroupResponseWithError, 0)
-	if response.Err != ErrUnknownMemberId {
+	if !errors.Is(response.Err, ErrUnknownMemberId) {
 		t.Error("Decoding error failed: ErrUnknownMemberId expected but found", response.Err)
 	}
 }
diff --git a/list_groups_response_test.go b/list_groups_response_test.go
index 1c21279..3f3112d 100644
--- a/list_groups_response_test.go
+++ b/list_groups_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"testing"
 )
 
@@ -29,7 +30,7 @@ func TestListGroupsResponse(t *testing.T) {
 
 	response = new(ListGroupsResponse)
 	testVersionDecodable(t, "no error", response, listGroupsResponseEmpty, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Expected no gerror, found:", response.Err)
 	}
 	if len(response.Groups) != 0 {
@@ -38,7 +39,7 @@ func TestListGroupsResponse(t *testing.T) {
 
 	response = new(ListGroupsResponse)
 	testVersionDecodable(t, "no error", response, listGroupsResponseError, 0)
-	if response.Err != ErrClusterAuthorizationFailed {
+	if !errors.Is(response.Err, ErrClusterAuthorizationFailed) {
 		t.Error("Expected no gerror, found:", response.Err)
 	}
 	if len(response.Groups) != 0 {
@@ -47,7 +48,7 @@ func TestListGroupsResponse(t *testing.T) {
 
 	response = new(ListGroupsResponse)
 	testVersionDecodable(t, "no error", response, listGroupsResponseWithConsumer, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Expected no gerror, found:", response.Err)
 	}
 	if len(response.Groups) != 1 {
diff --git a/metadata_response_test.go b/metadata_response_test.go
index 248d9b0..28d728c 100644
--- a/metadata_response_test.go
+++ b/metadata_response_test.go
@@ -1,6 +1,9 @@
 package sarama
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 var (
 	emptyMetadataResponseV0 = []byte{
@@ -161,7 +164,7 @@ func TestMetadataResponseWithTopicsV0(t *testing.T) {
 		t.Fatal("Decoding produced", len(response.Topics), "topics where there were two!")
 	}
 
-	if response.Topics[0].Err != ErrNoError {
+	if !errors.Is(response.Topics[0].Err, ErrNoError) {
 		t.Error("Decoding produced invalid topic 0 error.")
 	}
 
@@ -173,7 +176,7 @@ func TestMetadataResponseWithTopicsV0(t *testing.T) {
 		t.Fatal("Decoding produced invalid partition count for topic 0.")
 	}
 
-	if response.Topics[0].Partitions[0].Err != ErrInvalidMessageSize {
+	if !errors.Is(response.Topics[0].Partitions[0].Err, ErrInvalidMessageSize) {
 		t.Error("Decoding produced invalid topic 0 partition 0 error.")
 	}
 
@@ -198,7 +201,7 @@ func TestMetadataResponseWithTopicsV0(t *testing.T) {
 		t.Error("Decoding produced invalid topic 0 partition 0 isr length.")
 	}
 
-	if response.Topics[1].Err != ErrNoError {
+	if !errors.Is(response.Topics[1].Err, ErrNoError) {
 		t.Error("Decoding produced invalid topic 1 error.")
 	}
 
diff --git a/mocks/async_producer_test.go b/mocks/async_producer_test.go
index 666c93b..4b8e4b3 100644
--- a/mocks/async_producer_test.go
+++ b/mocks/async_producer_test.go
@@ -68,7 +68,7 @@ func TestProducerReturnsExpectationsToChannels(t *testing.T) {
 		t.Error("Expected message 2 to be returned second")
 	}
 
-	if err1.Msg.Topic != "test 3" || err1.Err != sarama.ErrOutOfBrokers {
+	if err1.Msg.Topic != "test 3" || !errors.Is(err1, sarama.ErrOutOfBrokers) {
 		t.Error("Expected message 3 to be returned as error")
 	}
 
diff --git a/mocks/consumer_test.go b/mocks/consumer_test.go
index 2efe183..d0ae277 100644
--- a/mocks/consumer_test.go
+++ b/mocks/consumer_test.go
@@ -1,6 +1,7 @@
 package mocks
 
 import (
+	"errors"
 	"sort"
 	"testing"
 
@@ -43,7 +44,7 @@ func TestConsumerHandlesExpectations(t *testing.T) {
 		t.Error("Message was not as expected:", test0_msg)
 	}
 	test0_err := <-pc_test0.Errors()
-	if test0_err.Err != sarama.ErrOutOfBrokers {
+	if !errors.Is(test0_err, sarama.ErrOutOfBrokers) {
 		t.Error("Expected sarama.ErrOutOfBrokers, found:", test0_err.Err)
 	}
 
@@ -96,7 +97,7 @@ func TestConsumerHandlesExpectationsPausingResuming(t *testing.T) {
 		t.Error("Problem to pause consumption")
 	}
 	test0_err := <-pc_test0.Errors()
-	if test0_err.Err != sarama.ErrOutOfBrokers {
+	if !errors.Is(test0_err, sarama.ErrOutOfBrokers) {
 		t.Error("Expected sarama.ErrOutOfBrokers, found:", test0_err.Err)
 	}
 
@@ -159,15 +160,18 @@ func TestConsumerReturnsNonconsumedErrorsOnClose(t *testing.T) {
 
 	select {
 	case <-pc.Messages():
-		t.Error("Did not epxect a message on the messages channel.")
+		t.Error("Did not expect a message on the messages channel.")
 	case err := <-pc.Errors():
-		if err.Err != sarama.ErrOutOfBrokers {
+		if !errors.Is(err, sarama.ErrOutOfBrokers) {
 			t.Error("Expected sarama.ErrOutOfBrokers, found", err)
 		}
 	}
 
-	errs := pc.Close().(sarama.ConsumerErrors)
-	if len(errs) != 1 && errs[0].Err != sarama.ErrOutOfBrokers {
+	var errs sarama.ConsumerErrors
+	if !errors.As(pc.Close(), &errs) {
+		t.Error("Expected Close to return ConsumerErrors")
+	}
+	if len(errs) != 1 && !errors.Is(errs[0], sarama.ErrOutOfBrokers) {
 		t.Error("Expected Close to return the remaining sarama.ErrOutOfBrokers")
 	}
 }
@@ -178,7 +182,7 @@ func TestConsumerWithoutExpectationsOnPartition(t *testing.T) {
 	consumer := NewConsumer(trm, NewTestConfig())
 
 	_, err := consumer.ConsumePartition("test", 1, sarama.OffsetOldest)
-	if err != errOutOfExpectations {
+	if !errors.Is(err, errOutOfExpectations) {
 		t.Error("Expected ConsumePartition to return errOutOfExpectations")
 	}
 
@@ -329,7 +333,7 @@ func TestConsumerUnexpectedTopicMetadata(t *testing.T) {
 	trm := newTestReporterMock()
 	consumer := NewConsumer(trm, NewTestConfig())
 
-	if _, err := consumer.Topics(); err != sarama.ErrOutOfBrokers {
+	if _, err := consumer.Topics(); !errors.Is(err, sarama.ErrOutOfBrokers) {
 		t.Error("Expected sarama.ErrOutOfBrokers, found", err)
 	}
 
diff --git a/mocks/sync_producer_test.go b/mocks/sync_producer_test.go
index 00b8763..a7b53d1 100644
--- a/mocks/sync_producer_test.go
+++ b/mocks/sync_producer_test.go
@@ -48,7 +48,7 @@ func TestSyncProducerReturnsExpectationsToSendMessage(t *testing.T) {
 	}
 
 	_, _, err = sp.SendMessage(msg)
-	if err != sarama.ErrOutOfBrokers {
+	if !errors.Is(err, sarama.ErrOutOfBrokers) {
 		t.Errorf("The third message should not have been produced successfully")
 	}
 
@@ -89,7 +89,7 @@ func TestSyncProducerWithTooFewExpectations(t *testing.T) {
 	if _, _, err := sp.SendMessage(msg); err != nil {
 		t.Error("No error expected on first SendMessage call", err)
 	}
-	if _, _, err := sp.SendMessage(msg); err != errOutOfExpectations {
+	if _, _, err := sp.SendMessage(msg); !errors.Is(err, errOutOfExpectations) {
 		t.Error("errOutOfExpectations expected on second SendMessage call, found:", err)
 	}
 
diff --git a/offset_manager_test.go b/offset_manager_test.go
index 36ea0a0..85d635f 100644
--- a/offset_manager_test.go
+++ b/offset_manager_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -90,7 +91,7 @@ func TestNewOffsetManager(t *testing.T) {
 	safeClose(t, testClient)
 
 	_, err = NewOffsetManagerFromClient("group", testClient)
-	if err != ErrClosedClient {
+	if !errors.Is(err, ErrClosedClient) {
 		t.Errorf("Error expected for closed client; actual value: %v", err)
 	}
 }
diff --git a/offset_response_test.go b/offset_response_test.go
index 8dcb3bf..550923b 100644
--- a/offset_response_test.go
+++ b/offset_response_test.go
@@ -1,6 +1,9 @@
 package sarama
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 var (
 	emptyOffsetResponse = []byte{
@@ -72,7 +75,7 @@ func TestNormalOffsetResponse(t *testing.T) {
 		t.Fatal("Decoding produced", len(response.Blocks["z"]), "partitions for topic 'z' where there was one.")
 	}
 
-	if response.Blocks["z"][2].Err != ErrNoError {
+	if !errors.Is(response.Blocks["z"][2].Err, ErrNoError) {
 		t.Fatal("Decoding produced invalid error for topic z partition 2.")
 	}
 
@@ -103,7 +106,7 @@ func TestNormalOffsetResponseV1(t *testing.T) {
 		t.Fatal("Decoding produced", len(response.Blocks["z"]), "partitions for topic 'z' where there was one.")
 	}
 
-	if response.Blocks["z"][2].Err != ErrNoError {
+	if !errors.Is(response.Blocks["z"][2].Err, ErrNoError) {
 		t.Fatal("Decoding produced invalid error for topic z partition 2.")
 	}
 
diff --git a/produce_response_test.go b/produce_response_test.go
index b989e5a..f3c28eb 100644
--- a/produce_response_test.go
+++ b/produce_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -87,7 +88,7 @@ func TestProduceResponseDecode(t *testing.T) {
 		if block == nil {
 			t.Error("Decoding did not produce a block for foo/1")
 		} else {
-			if block.Err != ErrInvalidMessage {
+			if !errors.Is(block.Err, ErrInvalidMessage) {
 				t.Error("Decoding failed for foo/1/Err, got:", int16(block.Err))
 			}
 			if block.Offset != 255 {
@@ -149,7 +150,8 @@ func TestProduceResponseEncodeInvalidTimestamp(t *testing.T) {
 	if err == nil {
 		t.Error("Expecting error, got nil")
 	}
-	if _, ok := err.(PacketEncodingError); !ok {
+	target := PacketEncodingError{}
+	if !errors.As(err, &target) {
 		t.Error("Expecting PacketEncodingError, got:", err)
 	}
 }
diff --git a/sasl_handshake_response_test.go b/sasl_handshake_response_test.go
index 780bcf5..6987867 100644
--- a/sasl_handshake_response_test.go
+++ b/sasl_handshake_response_test.go
@@ -1,6 +1,9 @@
 package sarama
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 var saslHandshakeResponse = []byte{
 	0x00, 0x00,
@@ -12,7 +15,7 @@ func TestSaslHandshakeResponse(t *testing.T) {
 	t.Parallel()
 	response := new(SaslHandshakeResponse)
 	testVersionDecodable(t, "no error", response, saslHandshakeResponse, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding error failed: no error expected but found", response.Err)
 	}
 	if response.EnabledMechanisms[0] != "foo" {
diff --git a/sync_group_response_test.go b/sync_group_response_test.go
index 85faec4..9d6a6fa 100644
--- a/sync_group_response_test.go
+++ b/sync_group_response_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"reflect"
 	"testing"
 )
@@ -23,7 +24,7 @@ func TestSyncGroupResponse(t *testing.T) {
 
 	response = new(SyncGroupResponse)
 	testVersionDecodable(t, "no error", response, syncGroupResponseNoError, 0)
-	if response.Err != ErrNoError {
+	if !errors.Is(response.Err, ErrNoError) {
 		t.Error("Decoding Err failed: no error expected but found", response.Err)
 	}
 	if !reflect.DeepEqual(response.MemberAssignment, []byte{0x01, 0x02, 0x03}) {
@@ -32,7 +33,7 @@ func TestSyncGroupResponse(t *testing.T) {
 
 	response = new(SyncGroupResponse)
 	testVersionDecodable(t, "no error", response, syncGroupResponseWithError, 0)
-	if response.Err != ErrRebalanceInProgress {
+	if !errors.Is(response.Err, ErrRebalanceInProgress) {
 		t.Error("Decoding Err failed: ErrRebalanceInProgress expected but found", response.Err)
 	}
 	if !reflect.DeepEqual(response.MemberAssignment, []byte{}) {
diff --git a/sync_producer_test.go b/sync_producer_test.go
index b59d872..4d0f7c0 100644
--- a/sync_producer_test.go
+++ b/sync_producer_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"errors"
 	"log"
 	"sync"
 	"testing"
@@ -175,7 +176,7 @@ func TestSyncProducerToNonExistingTopic(t *testing.T) {
 	broker.Returns(metadataResponse)
 
 	_, _, err = producer.SendMessage(&ProducerMessage{Topic: "unknown"})
-	if err != ErrUnknownTopicOrPartition {
+	if !errors.Is(err, ErrUnknownTopicOrPartition) {
 		t.Error("Uxpected ErrUnknownTopicOrPartition, found:", err)
 	}
 
@@ -208,7 +209,7 @@ func TestSyncProducerRecoveryWithRetriesDisabled(t *testing.T) {
 	prodNotLeader.AddTopicPartition("my_topic", 0, ErrNotLeaderForPartition)
 	leader1.Returns(prodNotLeader)
 	_, _, err = producer.SendMessage(&ProducerMessage{Topic: "my_topic", Value: StringEncoder(TestMessage)})
-	if err != ErrNotLeaderForPartition {
+	if !errors.Is(err, ErrNotLeaderForPartition) {
 		t.Fatal(err)
 	}