fix: pass generic errors to sink handler if preset

Trendyol · May 23, 2024 · 9e94aec · 9e94aec
1 parent aec5947
commit 9e94aec
Showing 1 changed file with 10 additions and 0 deletions.
diff --git a/kafka/producer/producer_batch.go b/kafka/producer/producer_batch.go
@@ -138,6 +138,7 @@ func (b *Batch) FlushMessages() {
 				b.handleMessageTooLargeError(e)
 				return
 			default:
+				b.handleResponseError(e)
 				logger.Log.Error("batch producer flush error %v", err)
 				return
 			}
@@ -180,6 +181,15 @@ func (b *Batch) handleWriteError(writeErrors kafka.WriteErrors) {
 	}
 }
 
+func (b *Batch) handleResponseError(err error) {
+	for _, msg := range b.messages {
+		b.sinkResponseHandler.OnError(&gKafka.SinkResponseHandlerContext{
+			Message: convertKafkaMessage(msg),
+			Err:     err,
+		})
+	}
+}
+
 func (b *Batch) handleResponseSuccess() {
 	for _, msg := range b.messages {
 		b.sinkResponseHandler.OnSuccess(&gKafka.SinkResponseHandlerContext{