draft: implement non transactional batch consuming

batch_consumer.go

@@ -9,10 +9,11 @@ import (
 type batchConsumer struct {
 	*base
 
-	consumeFn func([]Message) error
+	consumeFn func(*[]Message) error
 
 	messageGroupLimit    int
 	messageGroupDuration time.Duration
+	manuelRetry          bool
 }
 
 func newBatchConsumer(cfg *ConsumerConfig) (Consumer, error) {
@@ -26,11 +27,12 @@ func newBatchConsumer(cfg *ConsumerConfig) (Consumer, error) {
 		consumeFn:            cfg.BatchConfiguration.BatchConsumeFn,
 		messageGroupLimit:    cfg.BatchConfiguration.MessageGroupLimit,
 		messageGroupDuration: cfg.BatchConfiguration.MessageGroupDuration,
+		manuelRetry:          cfg.ManuelRetryEnabled,
 	}
 
 	if cfg.RetryEnabled {
 		c.base.setupCronsumer(cfg, func(message kcronsumer.Message) error {
-			return c.consumeFn([]Message{toMessage(message)})
+			return c.consumeFn(&[]Message{toMessage(message)})
 		})
 	}
 
@@ -89,15 +91,15 @@ func (b *batchConsumer) startBatch() {
 }
 
 func (b *batchConsumer) process(messages []Message) {
-	consumeErr := b.consumeFn(messages)
+	consumeErr := b.consumeFn(&messages)
 
 	if consumeErr != nil {
 		b.logger.Warnf("Consume Function Err %s, Messages will be retried", consumeErr.Error())
-		// Try to process same messages again
-		if consumeErr = b.consumeFn(messages); 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(messages))
+		// if manuel retry enabled
+		if b.manuelRetry {
+			b.retry(getUnsuccessfulMessages(messages))
 		}
+		b.retry(messages)
 
 		if consumeErr != nil && b.retryEnabled {
 			cronsumerMessages := make([]kcronsumer.Message, 0, len(messages))
@@ -115,3 +117,23 @@ func (b *batchConsumer) process(messages []Message) {
 		b.metric.TotalProcessedMessagesCounter += int64(len(messages))
 	}
 }
+
+func (b *batchConsumer) retry(messages []Message) {
+	// Try to process same messages again
+	if consumeErr := b.consumeFn(&messages); 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(messages))
+	}
+	return
+}
+
+func getUnsuccessfulMessages(messages []Message) []Message {
+	var unsuccessfulMessages []Message
+	for _, message := range messages {
+		if !message.isSuccessful {
+			unsuccessfulMessages = append(unsuccessfulMessages, message)
+		}
+	}
+
+	return unsuccessfulMessages
+}

consumer_config.go

@@ -15,7 +15,7 @@ import (
 
 type ReaderConfig kafka.ReaderConfig
 
-type BatchConsumeFn func([]Message) error
+type BatchConsumeFn func(*[]Message) error
 
 type ConsumeFn func(Message) error
 
@@ -44,6 +44,7 @@ type ConsumerConfig struct {
 	Concurrency                     int
 	RetryEnabled                    bool
 	APIEnabled                      bool
+	ManuelRetryEnabled              bool
 }
 
 func (cfg *ConsumerConfig) newCronsumerConfig() *kcronsumer.Config {

message.go

@@ -20,7 +20,8 @@ type Message struct {
 	WriterData    interface{}
 	Time          time.Time
 	// Context To enable distributed tracing support
-	Context context.Context
+	Context      context.Context
+	isSuccessful bool
 }
 
 func (m *Message) toKafkaMessage() kafka.Message {
@@ -49,6 +50,7 @@ func fromKafkaMessage(message *kafka.Message) Message {
 		WriterData:    message.WriterData,
 		Time:          message.Time,
 		Context:       context.TODO(),
+		isSuccessful: true,
 	}
 }