feat: add transactional retry feature

README.md

@@ -119,6 +119,45 @@ After running `docker-compose up` command, you can run any application you want.
     }
 </details>
 
+<details>
+    <summary>With Non-Transactional Retry</summary>
+
+    func main() {
+        consumerCfg := &kafka.ConsumerConfig{
+            Reader: kafka.ReaderConfig{
+                Brokers: []string{"localhost:29092"},
+                Topic:   "standart-topic",
+                GroupID: "standart-cg",
+            },
+            LogLevel:     kafka.LogLevelDebug,
+            RetryEnabled: true,
+            NonTransactionalBatchRetryEnabled: true,
+            RetryConfiguration: kafka.RetryConfiguration{
+                Brokers:       []string{"localhost:29092"},
+                Topic:         "retry-topic",
+                StartTimeCron: "*/1 * * * *",
+                WorkDuration:  50 * time.Second,
+                MaxRetry:      3,
+            },
+            BatchConfiguration: kafka.BatchConfiguration{
+                MessageGroupLimit:    1000,
+                MessageGroupDuration: time.Second,
+                BatchConsumeFn:       batchConsumeFn,
+            },
+        }
+
+        consumer, _ := kafka.NewConsumer(consumerCfg)
+        defer consumer.Stop()
+
+        consumer.Consume()
+    }
+
+    func batchConsumeFn(messages []kafka.Message) error {
+        fmt.Printf("%d\n comes first %s", len(messages), messages[0].Value)
+        return nil
+    }
+</details>
+
 
 #### With Distributed Tracing Support
 
@@ -147,6 +186,7 @@ under [the specified folder](examples/with-sasl-plaintext) and then start the ap
 | `logLevel`                                       | Describes log level; valid options are `debug`, `info`, `warn`, and `error`                                                           | info                        |
 | `concurrency`                                    | Number of goroutines used at listeners                                                                                                | 1                           |
 | `retryEnabled`                                   | Retry/Exception consumer is working or not                                                                                            | false                       |
+| `nonTransactionalBatchRetryEnabled`              | Manuel error handling for batch consumers                                                                                             | false                       |
 | `commitInterval`                                 | indicates the interval at which offsets are committed to the broker.                                                                  | 1s                          |
 | `rack`                                           | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go#RackAffinityGroupBalancer)                                                 |                             |
 | `clientId`                                       | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go@v0.4.42#Dialer)                                                            |                             |

batch_consumer.go

@@ -101,8 +101,16 @@ func (b *batchConsumer) process(messages []*Message) {
 
 		if consumeErr != nil && b.retryEnabled {
 			cronsumerMessages := make([]kcronsumer.Message, 0, len(messages))
-			for i := range messages {
-				cronsumerMessages = append(cronsumerMessages, messages[i].toRetryableMessage(b.retryTopic))
+			if b.nonTransactionalBatchRetryEnabled {
+				for i := range messages {
+					if messages[i].IsFailed {
+						cronsumerMessages = append(cronsumerMessages, messages[i].toRetryableMessage(b.retryTopic))
+					}
+				}
+			} else {
+				for i := range messages {
+					cronsumerMessages = append(cronsumerMessages, messages[i].toRetryableMessage(b.retryTopic))
+				}
 			}
 
 			if produceErr := b.base.cronsumer.ProduceBatch(cronsumerMessages); produceErr != nil {

batch_consumer_test.go

@@ -169,6 +169,33 @@ func Test_batchConsumer_process(t *testing.T) {
 	})
 }
 
+func Test_batchConsumer_process_ReprocessingFailedWithRetryEnabled(t *testing.T) {
+	// Given
+	mc := &mockCronsumer{wantErr: true}
+	bc := batchConsumer{
+		base: &base{
+			metric: &ConsumerMetric{}, logger: NewZapLogger(LogLevelDebug),
+			retryEnabled:                      true,
+			nonTransactionalBatchRetryEnabled: true,
+			cronsumer:                         mc,
+		},
+		consumeFn: func(messages []*Message) error {
+			return errors.New("error case")
+		},
+	}
+
+	// When
+	bc.process([]*Message{{IsFailed: true}, {IsFailed: true}, {}})
+
+	// Then
+	if bc.metric.TotalProcessedMessagesCounter != 0 {
+		t.Fatalf("Total Processed Message Counter must equal to 0")
+	}
+	if bc.metric.TotalUnprocessedMessagesCounter != 3 {
+		t.Fatalf("Total Unprocessed Message Counter must equal to 3")
+	}
+}
+
 type mockCronsumer struct {
 	wantErr bool
 }

consumer_base.go

@@ -31,23 +31,24 @@
 }
 
 type base struct {
-	cronsumer                 kcronsumer.Cronsumer
-	api                       API
-	logger                    LoggerInterface
-	metric                    *ConsumerMetric
-	context                   context.Context
-	messageCh                 chan *Message
-	quit                      chan struct{}
-	cancelFn                  context.CancelFunc
-	r                         Reader
-	retryTopic                string
-	subprocesses              subprocesses
-	wg                        sync.WaitGroup
-	concurrency               int
-	once                      sync.Once
-	retryEnabled              bool
-	distributedTracingEnabled bool
-	propagator                propagation.TextMapPropagator
+	cronsumer                         kcronsumer.Cronsumer
+	api                               API
+	logger                            LoggerInterface
+	metric                            *ConsumerMetric
+	context                           context.Context
+	messageCh                         chan *Message
+	quit                              chan struct{}
+	cancelFn                          context.CancelFunc
+	r                                 Reader
+	retryTopic                        string
+	subprocesses                      subprocesses
+	wg                                sync.WaitGroup
+	concurrency                       int
+	once                              sync.Once
+	retryEnabled                      bool
+	nonTransactionalBatchRetryEnabled bool
+	distributedTracingEnabled         bool
+	propagator                        propagation.TextMapPropagator
 }
 
 func NewConsumer(cfg *ConsumerConfig) (Consumer, error) {
@@ -68,15 +69,16 @@
 	}
 
 	c := base{
-		metric:                    &ConsumerMetric{},
-		messageCh:                 make(chan *Message, cfg.Concurrency),
-		quit:                      make(chan struct{}),
-		concurrency:               cfg.Concurrency,
-		retryEnabled:              cfg.RetryEnabled,
-		distributedTracingEnabled: cfg.DistributedTracingEnabled,
-		logger:                    log,
-		subprocesses:              newSubProcesses(),
-		r:                         reader,
+		metric:                            &ConsumerMetric{},
+		messageCh:                         make(chan *Message, cfg.Concurrency),
+		quit:                              make(chan struct{}),
+		concurrency:                       cfg.Concurrency,
+		retryEnabled:                      cfg.RetryEnabled,
+		nonTransactionalBatchRetryEnabled: cfg.NonTransactionalBatchRetryEnabled,
+		distributedTracingEnabled:         cfg.DistributedTracingEnabled,
+		logger:                            log,
+		subprocesses:                      newSubProcesses(),
+		r:                                 reader,
 	}
 
 	if cfg.DistributedTracingEnabled {

consumer_config.go

@@ -25,25 +25,26 @@ type DialConfig struct {
 }
 
 type ConsumerConfig struct {
-	APIConfiguration                APIConfiguration
-	Logger                          LoggerInterface
-	MetricConfiguration             MetricConfiguration
-	SASL                            *SASLConfig
-	TLS                             *TLSConfig
-	Dial                            *DialConfig
-	BatchConfiguration              *BatchConfiguration
-	ConsumeFn                       ConsumeFn
-	ClientID                        string
-	Rack                            string
-	LogLevel                        LogLevel
-	Reader                          ReaderConfig
-	RetryConfiguration              RetryConfiguration
-	CommitInterval                  time.Duration
-	DistributedTracingEnabled       bool
-	DistributedTracingConfiguration DistributedTracingConfiguration
-	Concurrency                     int
-	RetryEnabled                    bool
-	APIEnabled                      bool
+	APIConfiguration                  APIConfiguration
+	Logger                            LoggerInterface
+	MetricConfiguration               MetricConfiguration
+	SASL                              *SASLConfig
+	TLS                               *TLSConfig
+	Dial                              *DialConfig
+	BatchConfiguration                *BatchConfiguration
+	ConsumeFn                         ConsumeFn
+	ClientID                          string
+	Rack                              string
+	LogLevel                          LogLevel
+	Reader                            ReaderConfig
+	RetryConfiguration                RetryConfiguration
+	CommitInterval                    time.Duration
+	DistributedTracingEnabled         bool
+	DistributedTracingConfiguration   DistributedTracingConfiguration
+	Concurrency                       int
+	RetryEnabled                      bool
+	APIEnabled                        bool
+	NonTransactionalBatchRetryEnabled bool
 }
 
 func (cfg *ConsumerConfig) newCronsumerConfig() *kcronsumer.Config {

examples/with-kafka-batch-consumer/main.go

@@ -42,7 +42,7 @@ func main() {
 }
 
 // In order to load topic with data, use:
-// kafka-console-producer --broker-list localhost:29092 --topic standart-topic < examples/with-kafka-batch-consumer/load.txt
+// kafka-console-producer --broker-list localhost:29092 --topic standart-topic < examples/load.txt
 func batchConsumeFn(messages []*kafka.Message) error {
 	fmt.Printf("%d\n comes first %s", len(messages), messages[0].Value)
 	return nil

examples/with-kafka-non-transactional-retry/main.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"github.com/Trendyol/kafka-konsumer"
+	"os"
+	"os/signal"
+	"time"
+)
+
+func main() {
+	consumerCfg := &kafka.ConsumerConfig{
+		Reader: kafka.ReaderConfig{
+			Brokers: []string{"localhost:29092"},
+			Topic:   "standart-topic",
+			GroupID: "standart-cg",
+		},
+		BatchConfiguration: &kafka.BatchConfiguration{
+			MessageGroupLimit:    1000,
+			MessageGroupDuration: time.Second,
+			BatchConsumeFn:       batchConsumeFn,
+		},
+		RetryEnabled:                      true,
+		NonTransactionalBatchRetryEnabled: true,
+		RetryConfiguration: kafka.RetryConfiguration{
+			Brokers:       []string{"localhost:29092"},
+			Topic:         "retry-topic",
+			StartTimeCron: "*/1 * * * *",
+			WorkDuration:  50 * time.Second,
+			MaxRetry:      3,
+		},
+	}
+
+	consumer, _ := kafka.NewConsumer(consumerCfg)
+	defer consumer.Stop()
+
+	consumer.Consume()
+
+	fmt.Println("Consumer started...!")
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt)
+	<-c
+}
+
+// In order to load topic with data, use:
+// kafka-console-producer --broker-list localhost:29092 --topic standart-topic < examples/load.txt
+func batchConsumeFn(messages []*kafka.Message) error {
+	fmt.Printf("%d\n comes first %s", len(messages), messages[0].Value)
+	// you can add custom error handling here & flag messages
+	for i := range messages {
+		if i%2 == 0 {
+			messages[i].IsFailed = true
+		} else {
+			messages[i].IsFailed = false
+		}
+	}
+
+	// you should return error here to retry failed messages
+	return errors.New("err")
+}

message.go

@@ -16,6 +16,7 @@ type Message struct {
 	Partition     int
 	Offset        int64
 	HighWaterMark int64
+	IsFailed      bool
 	Key           []byte
 	Value         []byte
 	Headers       []Header