fix: Prevent infinite retry for all errors

kafka/producer/producer_batch.go

@@ -2,11 +2,8 @@ package producer
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"io"
 	"sync"
-	"syscall"
 	"time"
 
 	"github.com/Trendyol/go-dcp-kafka/couchbase"
@@ -125,21 +122,13 @@ func (b *Batch) FlushMessages() {
 		err := b.Writer.WriteMessages(context.Background(), b.messages...)
 
 		if err != nil && b.sinkResponseHandler == nil {
-			if isFatalError(err) {
-				err = fmt.Errorf("permanent error on kafka side %v", err)
-				logger.Log.Error("error while flush, err: %v", err)
-				panic(err)
-			}
-			logger.Log.Error("batch producer flush error %v", err)
-			return
+			err = fmt.Errorf("batch producer flush error %v", err)
+			logger.Log.Error("error while flush, err: %v", err)
+			panic(err)
 		}
 
 		b.metric.BatchProduceLatency = time.Since(startedTime).Milliseconds()
 
-		b.messages = b.messages[:0]
-		b.currentMessageBytes = 0
-		b.batchTicker.Reset(b.batchTickerDuration)
-
 		if b.sinkResponseHandler != nil {
 			switch e := err.(type) {
 			case nil:
@@ -148,30 +137,17 @@ func (b *Batch) FlushMessages() {
 				b.handleWriteError(e)
 			case kafka.MessageTooLargeError:
 				b.handleMessageTooLargeError(e)
-				return
 			default:
 				b.handleResponseError(e)
-				logger.Log.Error("batch producer flush error %v", err)
-				return
+				logger.Log.Error("batch producer flush default error %v", err)
 			}
 		}
-	}
-	b.dcpCheckpointCommit()
-}
 
-func isFatalError(err error) bool {
-	e, ok := err.(kafka.Error)
-	if ok && errors.Is(err, kafka.UnknownTopicOrPartition) {
-		return true
-	}
-	if (ok && e.Temporary()) ||
-		errors.Is(err, io.ErrUnexpectedEOF) ||
-		errors.Is(err, syscall.ECONNREFUSED) ||
-		errors.Is(err, syscall.ECONNRESET) ||
-		errors.Is(err, syscall.EPIPE) {
-		return false
+		b.messages = b.messages[:0]
+		b.currentMessageBytes = 0
+		b.batchTicker.Reset(b.batchTickerDuration)
 	}
-	return true
+	b.dcpCheckpointCommit()
 }
 
 func (b *Batch) handleWriteError(writeErrors kafka.WriteErrors) {