refactor: migrate to auto commit

README.md

@@ -185,10 +185,8 @@ Kafka Konsumer offers an API that handles exposing several metrics.
 | Metric Name | Description | Value Type |
 |---------------------------------------------------------|---------------------------------------------|------------|
 | kafka_konsumer_processed_messages_total_current | Total number of processed messages. | Counter |
-| kafka_konsumer_processed_batch_messages_total_current | Total number of processed batch messages. | Counter |
 | kafka_konsumer_unprocessed_messages_total_current | Total number of unprocessed messages. | Counter |
+| kafka_konsumer_processed_batch_messages_total_current | Total number of processed batch messages. | Counter |
 | kafka_konsumer_unprocessed_batch_messages_total_current | Total number of unprocessed batch messages. | Counter |
 
-**NOTE:** `kafka_konsumer_processed_batch_messages_total_current` and `kafka_konsumer_unprocessed_batch_messages_total_current` will
-be deprecated in the next releases. 
-Please use `kafka_konsumer_processed_messages_total_current` and `kafka_konsumer_unprocessed_messages_total_current` instead.
+**NOTE:** `kafka_konsumer_processed_batch_messages_total_current` and `kafka_konsumer_unprocessed_batch_messages_total_current` will be deprecated in the next releases. Please use `kafka_konsumer_processed_messages_total_current` and `kafka_konsumer_unprocessed_messages_total_current` instead.

batch_consumer.go

@@ -4,7 +4,6 @@ import (
  "time"
 
  kcronsumer "github.com/Trendyol/kafka-cronsumer/pkg/kafka"
- "github.com/segmentio/kafka-go"
 )
 
 type batchConsumer struct {
@@ -55,8 +54,6 @@ func (b *batchConsumer) Consume() {
  b.wg.Add(1)
  go b.startBatch()
  }
-
- go b.handleCommit()
 }
 
 func (b *batchConsumer) startBatch() {
@@ -98,6 +95,8 @@ func (b *batchConsumer) process(messages []Message) {
 
  // Try to process same message again
  if consumeErr = b.consumeFn(messages); consumeErr != nil {
+ b.metric.TotalUnprocessedMessagesCounter += int64(len(messages))
+
  b.logger.Warnf("Consume Function Again Err %s, messages are sending to exception/retry topic %s", consumeErr.Error(), b.retryTopic)
 
  cronsumerMessages := make([]kcronsumer.Message, 0, len(messages))
@@ -111,10 +110,5 @@ func (b *batchConsumer) process(messages []Message) {
  }
  }
 
- segmentioMessages := make([]kafka.Message, 0, len(messages))
- for i := range messages {
- segmentioMessages = append(segmentioMessages, kafka.Message(messages[i]))
- }
-
- b.commitReq <- segmentioMessages
+ b.metric.TotalProcessedMessagesCounter += int64(len(messages))
 }

consumer.go

@@ -2,7 +2,6 @@ package kafka
 
 import (
  kcronsumer "github.com/Trendyol/kafka-cronsumer/pkg/kafka"
- "github.com/segmentio/kafka-go"
 )
 
 type consumer struct {
@@ -50,8 +49,6 @@ func (c *consumer) Consume() {
  }
  }()
  }
-
- go c.handleCommit()
 }
 
 func (c *consumer) process(message Message) {
@@ -61,6 +58,7 @@ func (c *consumer) process(message Message) {
 
  // Try to process same message again
  if consumeErr = c.consumeFn(message); consumeErr != nil {
+ c.metric.TotalUnprocessedMessagesCounter++
  c.logger.Warnf("Consume Function Again Err %s, message is sending to exception/retry topic %s", consumeErr.Error(), c.retryTopic)
 
  retryableMsg := message.toRetryableMessage(c.retryTopic)
@@ -71,5 +69,5 @@ func (c *consumer) process(message Message) {
  }
  }
 
- c.commitReq <- []kafka.Message{kafka.Message(message)}
+ c.metric.TotalProcessedMessagesCounter++
 }

consumer_base.go

@@ -25,7 +25,6 @@ type base struct {
  context context.Context
  messageCh chan Message
  quit chan struct{}
- commitReq chan []kafka.Message
  cancelFn context.CancelFunc
  r *kafka.Reader
  retryTopic string
@@ -57,7 +56,6 @@ func newBase(cfg *ConsumerConfig) (*base, error) {
  metric: &ConsumerMetric{},
  messageCh: make(chan Message, cfg.Concurrency),
  quit: make(chan struct{}),
- commitReq: make(chan []kafka.Message),
  concurrency: cfg.Concurrency,
  retryEnabled: cfg.RetryEnabled,
  logger: log,
@@ -97,7 +95,7 @@ func (c *base) startConsume() {
  case <-c.quit:
  return
  default:
- message, err := c.r.FetchMessage(c.context)
+ message, err := c.r.ReadMessage(c.context)
  if err != nil {
  if c.context.Err() != nil {
  continue
@@ -111,33 +109,6 @@ func (c *base) startConsume() {
  }
 }
 
-func (c *base) handleCommit() {
- // it is used for tracking the latest committed offsets by topic => partition => offset
- offsets := offsetStash{}
-
- for msgs := range c.commitReq {
- // Extract messages which needed to commit
- willBeCommitted := offsets.IgnoreAlreadyCommittedMessages(msgs)
- if len(willBeCommitted) == 0 {
- continue
- }
-
- commitErr := c.r.CommitMessages(context.Background(), willBeCommitted...)
- if commitErr != nil {
- c.logger.Error("Error Committing messages %s", commitErr.Error())
- c.metric.TotalUnprocessedMessagesCounter++
- continue
- }
-
- // Update the latest offsets with recently committed messages
- offsets.Update(willBeCommitted)
-
- c.metric.TotalProcessedMessagesCounter++
-
- c.logger.Debug(offsets)
- }
-}
-
 func (c *base) WithLogger(logger LoggerInterface) {
  c.logger = logger
 }
@@ -151,7 +122,6 @@ func (c *base) Stop() error {
  c.quit <- struct{}{}
  close(c.messageCh)
  c.wg.Wait()
- close(c.commitReq)
  err = c.r.Close()
  })