feat: disable instant retry for non transactional retry mode

Trendyol · Jul 30, 2024 · 10830c3 · 10830c3
1 parent ff0c6b1
commit 10830c3
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 10 deletions.
diff --git a/batch_consumer.go b/batch_consumer.go
@@ -213,10 +213,14 @@ func (b *batchConsumer) process(chunkMessages []*Message) {
 	consumeErr := b.consumeFn(chunkMessages)
 
 	if consumeErr != nil {
-		b.logger.Warnf("Consume Function Err %s, Messages will be retried", consumeErr.Error())
-		// Try to process same messages again for resolving transient network errors etc.
-		if consumeErr = b.consumeFn(chunkMessages); consumeErr != nil {
-			b.logger.Warnf("Consume Function Again Err %s, messages are sending to exception/retry topic %s", consumeErr.Error(), b.retryTopic)
+		if b.transactionalRetry {
+			b.logger.Warnf("Consume Function Err %s, Messages will be retried", consumeErr.Error())
+			// Try to process same messages again for resolving transient network errors etc.
+			if consumeErr = b.consumeFn(chunkMessages); consumeErr != nil {
+				b.logger.Warnf("Consume Function Again Err %s, messages are sending to exception/retry topic %s", consumeErr.Error(), b.retryTopic)
+				b.metric.TotalUnprocessedMessagesCounter += int64(len(chunkMessages))
+			}
+		} else {
 			b.metric.TotalUnprocessedMessagesCounter += int64(len(chunkMessages))
 		}
 

diff --git a/consumer.go b/consumer.go
@@ -137,10 +137,14 @@ func (c *consumer) process(message *Message) {
 	consumeErr := c.consumeFn(message)
 
 	if consumeErr != nil {
-		c.logger.Warnf("Consume Function Err %s, Message will be retried", consumeErr.Error())
-		// Try to process same message again
-		if consumeErr = c.consumeFn(message); consumeErr != nil {
-			c.logger.Warnf("Consume Function Again Err %s, message is sending to exception/retry topic %s", consumeErr.Error(), c.retryTopic)
+		if c.transactionalRetry {
+			c.logger.Warnf("Consume Function Err %s, Message will be retried", consumeErr.Error())
+			// Try to process same message again
+			if consumeErr = c.consumeFn(message); consumeErr != nil {
+				c.logger.Warnf("Consume Function Again Err %s, message is sending to exception/retry topic %s", consumeErr.Error(), c.retryTopic)
+				c.metric.TotalUnprocessedMessagesCounter++
+			}
+		} else {
 			c.metric.TotalUnprocessedMessagesCounter++
 		}
 	}

diff --git a/consumer_test.go b/consumer_test.go
@@ -32,7 +32,7 @@ func Test_consumer_process(t *testing.T) {
 		// Given
 		gotOnlyOneTimeException := true
 		c := consumer{
-			base: &base{metric: &ConsumerMetric{}, logger: NewZapLogger(LogLevelDebug)},
+			base: &base{metric: &ConsumerMetric{}, logger: NewZapLogger(LogLevelDebug), transactionalRetry: true},
 			consumeFn: func(*Message) error {
 				if gotOnlyOneTimeException {
 					gotOnlyOneTimeException = false
@@ -47,7 +47,7 @@ func Test_consumer_process(t *testing.T) {
 
 		// Then
 		if c.metric.TotalProcessedMessagesCounter != 1 {
-			t.Fatalf("Total Processed Message Counter must equal to 3")
+			t.Fatalf("Total Processed Message Counter must equal to 1")
 		}
 		if c.metric.TotalUnprocessedMessagesCounter != 0 {
 			t.Fatalf("Total Unprocessed Message Counter must equal to 0")