Merge pull request #59 from Trendyol/feature/#55

feat: add error description to message header

Author: A.Samet İleri

internal/consumer.go

@@ -9,7 +9,7 @@ import (
 )
 
 type Consumer interface {
-	ReadMessage(ctx context.Context) (*MessageWrapper, error)
+	ReadMessage(ctx context.Context) (*segmentio.Message, error)
 	Stop()
 }
 
@@ -53,7 +53,7 @@ func newConsumer(kafkaConfig *kafka.Config) *kafkaConsumer {
 	}
 }
 
-func (k kafkaConsumer) ReadMessage(ctx context.Context) (*MessageWrapper, error) {
+func (k kafkaConsumer) ReadMessage(ctx context.Context) (*segmentio.Message, error) {
 	msg, err := k.consumer.ReadMessage(ctx)
 	if err != nil {
 		if isContextCancelled(err) {
@@ -63,7 +63,7 @@ func (k kafkaConsumer) ReadMessage(ctx context.Context) (*MessageWrapper, error)
 		return nil, err
 	}
 
-	return NewMessageWrapper(msg), nil
+	return &msg, nil
 }
 
 func isContextCancelled(err error) bool {

internal/cronsumer.go

@@ -48,16 +48,20 @@ func (k *kafkaCronsumer) Listen(ctx context.Context, strategyName string, cancel
 	startTime := time.Now()
 	startTimeUnixNano := startTime.UnixNano()
 
+	retryStrategy := kafka.GetBackoffStrategy(strategyName)
+
 	for {
-		msg, err := k.kafkaConsumer.ReadMessage(ctx)
+		m, err := k.kafkaConsumer.ReadMessage(ctx)
 		if err != nil {
-			k.cfg.Logger.Errorf("Message could not read, error %v", err)
+			k.cfg.Logger.Warnf("Message could not read, error %v", err)
 			return
 		}
-		if msg == nil {
+		if m == nil {
 			return
 		}
 
+		msg := NewMessageWrapper(*m, strategyName)
+
 		if msg.ProduceTime >= startTimeUnixNano {
 			(*cancelFuncWrapper)()
 
@@ -70,8 +74,6 @@ func (k *kafkaCronsumer) Listen(ctx context.Context, strategyName string, cancel
 			return
 		}
 
-		retryStrategy := kafka.GetBackoffStrategy(strategyName)
-
 		if retryStrategy.String() == kafka.FixedBackOffStrategy {
 			k.sendToMessageChannel(*msg)
 			continue
@@ -105,6 +107,7 @@ func (k *kafkaCronsumer) GetMetric() *CronsumerMetric {
 func (k *kafkaCronsumer) processMessage() {
 	for msg := range k.messageChannel {
 		if err := k.consumeFn(msg.Message); err != nil {
+			msg.AddHeader(createErrHeader(err))
 			k.produce(msg)
 		}
 	}

internal/cronsumer_test.go

@@ -196,8 +196,8 @@ type mockConsumer struct{}
 func (c mockConsumer) Stop() {
 }
 
-func (c mockConsumer) ReadMessage(ctx context.Context) (*MessageWrapper, error) {
-	return &MessageWrapper{}, nil
+func (c mockConsumer) ReadMessage(ctx context.Context) (*segmentio.Message, error) {
+	return &segmentio.Message{}, nil
 }
 
 type mockProducer struct {

internal/message.go

@@ -15,6 +15,7 @@ const (
 	RetryHeaderKey              = "x-retry-count"
 	RetryAttemptHeaderKey       = "x-retry-attempt-count"
 	MessageProduceTimeHeaderKey = "x-produce-time"
+	MessageErrHeaderKey         = "x-error-message"
 )
 
 type MessageWrapper struct {
@@ -24,22 +25,29 @@ type MessageWrapper struct {
 	RetryAttemptCount int
 }
 
-func NewMessageWrapper(msg segmentio.Message) *MessageWrapper {
-	return &MessageWrapper{
-		RetryCount:        getRetryCount(&msg),
-		RetryAttemptCount: getRetryAttemptCount(&msg),
-		ProduceTime:       getMessageProduceTime(&msg),
-		Message: kafka.Message{
-			Topic:         msg.Topic,
-			Partition:     msg.Partition,
-			Offset:        msg.Offset,
-			HighWaterMark: msg.HighWaterMark,
-			Key:           msg.Key,
-			Value:         msg.Value,
-			Headers:       FromHeaders(msg.Headers),
-			Time:          msg.Time,
-		},
+func NewMessageWrapper(msg segmentio.Message, strategyName string) *MessageWrapper {
+	mw := &MessageWrapper{
+		RetryCount:  getRetryCount(&msg),
+		ProduceTime: getMessageProduceTime(&msg),
 	}
+
+	// Don't add x-retry-attempt-count for fixed strategy.
+	if strategyName != kafka.FixedBackOffStrategy {
+		mw.RetryAttemptCount = getRetryAttemptCount(&msg)
+	}
+
+	mw.Message = kafka.Message{
+		Topic:         msg.Topic,
+		Partition:     msg.Partition,
+		Offset:        msg.Offset,
+		HighWaterMark: msg.HighWaterMark,
+		Key:           msg.Key,
+		Value:         msg.Value,
+		Headers:       FromHeaders(msg.Headers),
+		Time:          msg.Time,
+	}
+
+	return mw
 }
 
 func (m *MessageWrapper) To(increaseRetry bool, increaseRetryAttempt bool) segmentio.Message {

internal/message_header.go

@@ -33,6 +33,13 @@ func FromHeaders(sh []segmentio.Header) []kafka.Header {
 	return r
 }
 
+func createErrHeader(consumeErr error) kafka.Header {
+	return kafka.Header{
+		Key:   MessageErrHeaderKey,
+		Value: []byte(consumeErr.Error()),
+	}
+}
+
 func getRetryCount(message *segmentio.Message) int {
 	for i := range message.Headers {
 		if message.Headers[i].Key != RetryHeaderKey {

internal/message_header_test.go

@@ -2,6 +2,7 @@ package internal
 
 import (
 	"bytes"
+	"errors"
 	"strconv"
 	"testing"
 
@@ -190,3 +191,20 @@ func Test_getRetryAttempt(t *testing.T) {
 		}
 	})
 }
+
+func TestMessage_CreateErrHeader(t *testing.T) {
+	// Given
+	e := errors.New("err")
+
+	// When
+	h := createErrHeader(e)
+
+	// Then
+	if h.Key != MessageErrHeaderKey {
+		t.Fatalf("Header key must be equal to X-ErrMessage")
+	}
+
+	if h.Value == nil {
+		t.Fatalf("Header value must be present")
+	}
+}

internal/message_test.go

@@ -26,18 +26,19 @@ func Test_NewMessageWrapper(t *testing.T) {
 		WriterData: "1",
 		Time:       time.Now(),
 	}
+
 	// When
-	actual := NewMessageWrapper(expected)
+	actual := NewMessageWrapper(expected, FixedBackOffStrategy)
 	actualHeader := actual.Headers[0]
 	expectedHeader := expected.Headers[0]
+
 	// Then
 	if actual.Topic != expected.Topic {
 		t.Errorf("Expected: %s, Actual: %s", expected.Topic, actual.Topic)
 	}
 	if actual.Partition != expected.Partition {
 		t.Errorf("Expected: %d, Actual: %d", expected.Partition, actual.Partition)
 	}
-
 	if actual.Offset != expected.Offset {
 		t.Errorf("Expected: %d, Actual: %d", expected.Offset, actual.Offset)
 	}

pkg/kafka/cronsumer_message.go

@@ -61,6 +61,18 @@ func (mb *MessageBuilder) WithHighWatermark(highWaterMark int64) *MessageBuilder
 	return mb
 }
 
+// AddHeader works as a idempotent function
+func (m *Message) AddHeader(header Header) {
+	for i := range m.Headers {
+		if m.Headers[i].Key == header.Key {
+			m.Headers[i].Value = header.Value
+			return
+		}
+	}
+
+	m.Headers = append(m.Headers, header)
+}
+
 func (mb *MessageBuilder) Build() Message {
 	m := Message{}
 

pkg/kafka/cronsumer_message_test.go

@@ -129,3 +129,52 @@ func Test_WithHighWatermark(t *testing.T) {
 		t.Errorf("Expected: %d, Actual: %d", expected, *actual)
 	}
 }
+
+func TestMessage_AddHeader(t *testing.T) {
+	t.Run("When_New_Header_Comes", func(t *testing.T) {
+		// Given
+		m := Message{
+			Headers: []Header{
+				{Key: "foo", Value: []byte("fooValue")},
+			},
+		}
+
+		// When
+		m.AddHeader(Header{Key: "bar", Value: []byte("barValue")})
+
+		// Then
+		headers := m.Headers
+		if len(headers) != 2 {
+			t.Fatalf("Header length must be equal to 2")
+		}
+		if headers[1].Key != "bar" {
+			t.Fatalf("Header key must be equal to bar")
+		}
+		if !bytes.Equal(headers[1].Value, []byte("barValue")) {
+			t.Fatalf("Header value must be equal to barValue")
+		}
+	})
+	t.Run("When_Same_Header_Comes", func(t *testing.T) {
+		// Given
+		m := Message{
+			Headers: []Header{
+				{Key: "foo", Value: []byte("fooValue")},
+			},
+		}
+
+		// When
+		m.AddHeader(Header{Key: "foo", Value: []byte("barValue")})
+
+		// Then
+		headers := m.Headers
+		if len(headers) != 1 {
+			t.Fatalf("Header length must be equal to 1")
+		}
+		if headers[0].Key != "foo" {
+			t.Fatalf("Header key must be equal to foo")
+		}
+		if !bytes.Equal(headers[0].Value, []byte("barValue")) {
+			t.Fatalf("Header value must be equal to barValue")
+		}
+	})
+}