feat: throw panic when produce message to retry topic is unsuccessful

consumer.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -152,8 +153,10 @@ func (c *consumer) process(message *Message) {
 	if consumeErr != nil && c.retryEnabled {
 		retryableMsg := message.toRetryableMessage(c.retryTopic, consumeErr.Error())
 		if produceErr := c.cronsumer.Produce(retryableMsg); produceErr != nil {
-			c.logger.Errorf("Error producing message %s to exception/retry topic %s",
+			errorMessage := fmt.Sprintf("Error producing message %s to exception/retry topic %s",
 				string(retryableMsg.Value), produceErr.Error())
+			c.logger.Error(errorMessage)
+			panic(errorMessage)
 		}
 	}
 

consumer_test.go

@@ -104,16 +104,14 @@ func Test_consumer_process(t *testing.T) {
 			},
 		}
 
-		// When
-		c.process(&Message{})
+		defer func() {
+			if r := recover(); r == nil {
+				t.Errorf("The code did not panic")
+			}
+		}()
 
-		// Then
-		if c.metric.TotalProcessedMessagesCounter != 0 {
-			t.Fatalf("Total Processed Message Counter must equal to 0")
-		}
-		if c.metric.TotalUnprocessedMessagesCounter != 1 {
-			t.Fatalf("Total Unprocessed Message Counter must equal to 1")
-		}
+		// When && Then
+		c.process(&Message{})
 	})
 }