fix(kafka): simple retry logic fetch msg

messaging/reader.go

@@ -47,11 +47,26 @@ type KafkaReader struct {
 // readBroker us as a wrapper for kafka.Reader implementation to fulfill BrokerReader interface
 type readBroker struct {
 	*kafka.Reader
+	maxRetries      int
+	retriesInterval time.Duration
 }
 
 // FetchMessages used to fetch messages from the broker
 func (rm *readBroker) FetchMessage(ctx context.Context) (Message, error) {
-	m, err := rm.Reader.FetchMessage(ctx)
+	var err error
+	var m kafka.Message
+	var retryNumber = 0
+
+	for retryNumber <= rm.maxRetries {
+		m, err = rm.Reader.FetchMessage(ctx)
+		if err != nil {
+			retryNumber = retryNumber + 1
+			log.Errorf("# messaging # retry number %v failed, trying again, err: %v", retryNumber, err)
+			time.Sleep(rm.retriesInterval)
+			continue
+		}
+		break
+	}
 
 	if err != nil {
 		return Message{}, err
@@ -110,7 +125,7 @@ func NewReader(brokers []string, groupID string, topic string) *KafkaReader {
 	return &KafkaReader{brokers: brokers,
 		groupID:         groupID,
 		topic:           topic,
-		brokerReader:    &readBroker{kafkaReader},
+		brokerReader:    &readBroker{kafkaReader, retries, intervalTime},
 		maxRetries:      retries,
 		retriesInterval: intervalTime,
 	}

messaging/reader_test.go

@@ -3,10 +3,11 @@ package messaging
 import (
 	"context"
 	"fmt"
-	"github.com/microdevs/missy/service"
 	"testing"
 	"time"
 
+	"github.com/microdevs/missy/service"
+
 	"reflect"
 
 	"bou.ke/monkey"
@@ -198,7 +199,7 @@ func TestReadBroker_FetchMessage(t *testing.T) {
 
 	defer monkey.Unpatch(kr.FetchMessage)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	msg, err := rb.FetchMessage(context.Background())
 
@@ -256,7 +257,7 @@ func TestReadBroker_FetchMessage_Error(t *testing.T) {
 
 	defer monkey.Unpatch(kr.FetchMessage)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 1, 1}
 
 	_, err := rb.FetchMessage(context.Background())
 
@@ -287,7 +288,7 @@ func TestReadBroker_ReadMessage(t *testing.T) {
 
 	defer monkey.Unpatch(kr.ReadMessage)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	msg, err := rb.ReadMessage(context.Background())
 
@@ -334,7 +335,7 @@ func TestReadBroker_ReadMessage_Error(t *testing.T) {
 
 	defer monkey.Unpatch(kr.ReadMessage)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	_, err := rb.ReadMessage(context.Background())
 
@@ -371,7 +372,7 @@ func TestReadBroker_CommitMessages(t *testing.T) {
 
 	defer monkey.Unpatch(kr.CommitMessages)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	err := rb.CommitMessages(context.Background(), messages...)
 
@@ -409,7 +410,7 @@ func TestReadBroker_CommitMessages_Error(t *testing.T) {
 
 	defer monkey.Unpatch(kr.CommitMessages)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	err := rb.CommitMessages(context.Background(), messages...)
 
@@ -441,7 +442,7 @@ func TestReadBroker_Close(t *testing.T) {
 
 	defer monkey.Unpatch(kr.Close)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	err := rb.Close()
 
@@ -473,7 +474,7 @@ func TestReadBroker_Close_Error(t *testing.T) {
 
 	defer monkey.Unpatch(kr.Close)
 
-	rb := readBroker{kr}
+	rb := readBroker{kr, 0, 0}
 
 	err := rb.Close()