Add permanent error check (Trendyol#55)

* Add permanent error check

kafka/producer/producer_batch.go

@@ -3,7 +3,11 @@ package producer
 import (
 	"context"
 	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/Trendyol/go-dcp-client/logger"
@@ -85,6 +89,9 @@ func (b *producerBatch) FlushMessages() {
 		startedTime := time.Now()
 		err := b.Writer.WriteMessages(context.Background(), b.messages...)
 		if err != nil {
+			if isFatalError(err) {
+				panic(fmt.Errorf("permanent error on Kafka side %e", err))
+			}
 			b.errorLogger.Printf("batch producer flush error %v", err)
 			return
 		}
@@ -96,3 +103,16 @@ func (b *producerBatch) FlushMessages() {
 	}
 	b.dcpCheckpointCommit()
 }
+
+func isFatalError(err error) bool {
+	e, ok := err.(kafka.Error)
+
+	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
+	}
+	return true
+}