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)
+	})
 }