More Retryable Errors (#332)

lib/kafkalib/errors.go

@@ -1,6 +1,12 @@
 package kafkalib
 
-import "strings"
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/segmentio/kafka-go"
+)
 
 func isExceedMaxMessageBytesErr(err error) bool {
 	return err != nil && strings.Contains(err.Error(),
@@ -10,5 +16,20 @@ func isExceedMaxMessageBytesErr(err error) bool {
 // isRetryableError - returns true if the error is retryable
 // If it's retryable, you need to reload the Kafka client.
 func isRetryableError(err error) bool {
-	return err != nil && strings.Contains(err.Error(), "Topic Authorization Failed: the client is not authorized to access the requested topic")
+	if err == nil {
+		return false
+	}
+
+	retryableErrs := []error{
+		context.DeadlineExceeded,
+		kafka.TopicAuthorizationFailed,
+	}
+
+	for _, retryableErr := range retryableErrs {
+		if errors.Is(err, retryableErr) {
+			return true
+		}
+	}
+
+	return false
 }

lib/kafkalib/errors_test.go

@@ -1,7 +1,9 @@
 package kafkalib
 
 import (
+	"context"
 	"fmt"
+	"github.com/segmentio/kafka-go"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -31,3 +33,34 @@ func TestIsExceedMaxMessageBytesErr(t *testing.T) {
 		assert.Equal(t, tc.expected, actual, tc.err)
 	}
 }
+
+func TestIsRetryableError(t *testing.T) {
+	type _tc struct {
+		err      error
+		expected bool
+	}
+
+	tcs := []_tc{
+		{
+			err:      fmt.Errorf(""),
+			expected: false,
+		},
+		{
+			err:      nil,
+			expected: false,
+		},
+		{
+			err:      kafka.TopicAuthorizationFailed,
+			expected: true,
+		},
+		{
+			err:      context.DeadlineExceeded,
+			expected: true,
+		},
+	}
+
+	for _, tc := range tcs {
+		actual := isRetryableError(tc.err)
+		assert.Equal(t, tc.expected, actual, tc.err)
+	}
+}

lib/kafkalib/writer.go

@@ -4,19 +4,18 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"log/slog"
-	"time"
-
-	"github.com/artie-labs/transfer/lib/jitter"
-	"github.com/artie-labs/transfer/lib/size"
 	awsCfg "github.com/aws/aws-sdk-go-v2/config"
-	"github.com/segmentio/kafka-go"
 	"github.com/segmentio/kafka-go/sasl/aws_msk_iam_v2"
+	"log/slog"
+	"time"
 
 	"github.com/artie-labs/reader/config"
 	"github.com/artie-labs/reader/lib"
 	"github.com/artie-labs/reader/lib/iterator"
 	"github.com/artie-labs/reader/lib/mtr"
+	"github.com/artie-labs/transfer/lib/jitter"
+	"github.com/artie-labs/transfer/lib/size"
+	"github.com/segmentio/kafka-go"
 )
 
 const (