Add ErrorType method to MessageTooLargeError (segmentio#1311)

* Add ErrorType method to MessageTooLargeError struct to support functionality with the errors.Is function.

See segmentio#1293

* implement the unwrap method to satisfy errors.Is

* support error interface for unwrap

error.go

@@ -636,6 +636,7 @@ func coalesceErrors(errs ...error) error {
  return nil
 }
 
+// MessageTooLargeError is returned when a message is too large to fit within the allowed size.
 type MessageTooLargeError struct {
  Message Message
  Remaining []Message
@@ -655,6 +656,10 @@ func (e MessageTooLargeError) Error() string {
  return MessageSizeTooLarge.Error()
 }
 
+func (e MessageTooLargeError) Unwrap() error {
+ return MessageSizeTooLarge
+}
+
 func makeError(code int16, message string) error {
  if code == 0 {
  return nil

error_test.go

@@ -3,6 +3,8 @@ package kafka
 import (
  "fmt"
  "testing"
+
+ "github.com/stretchr/testify/assert"
 )
 
 func TestError(t *testing.T) {
@@ -110,4 +112,16 @@ func TestError(t *testing.T) {
  t.Error("non-empty description:", s)
  }
  })
+
+ t.Run("MessageTooLargeError error.Is satisfaction", func(t *testing.T) {
+ err := MessageSizeTooLarge
+ msg := []Message{
+ {Key: []byte("key"), Value: []byte("value")},
+ {Key: []byte("key"), Value: make([]byte, 8)},
+ }
+ msgTooLarge := messageTooLarge(msg, 1)
+ assert.NotErrorIs(t, err, msgTooLarge)
+ assert.Contains(t, msgTooLarge.Error(), MessageSizeTooLarge.Error())
+ assert.ErrorIs(t, msgTooLarge, MessageSizeTooLarge)
+ })
 }