bugfix #20

examples/pprof/config.yml

@@ -0,0 +1,10 @@
+brokers:
+  - "localhost:9092"
+consumer:
+  groupId: "sample-consumer"
+  topic: "exception"
+  maxRetry: 3
+  concurrency: 100
+  cron: "*/1 * * * *"
+  duration: 30s
+logLevel: debug

examples/pprof/main.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"fmt"
+	cronsumer "github.com/Trendyol/kafka-cronsumer"
+	"github.com/Trendyol/kafka-cronsumer/pkg/kafka"
+	"log"
+	"net/http"
+	"os"
+	"path/filepath"
+	"runtime"
+
+	"gopkg.in/yaml.v3"
+	_ "net/http/pprof"
+)
+
+func main() {
+	kafkaConfig := getConfig()
+
+	var consumeFn kafka.ConsumeFn = func(message kafka.Message) error {
+		fmt.Printf("consumer > Message received: %s\n", string(message.Value))
+		return nil
+	}
+
+	go func() {
+		log.Println(http.ListenAndServe("localhost:6060", nil))
+	}()
+
+	c := cronsumer.New(kafkaConfig, consumeFn)
+	c.Run()
+}
+
+func getConfig() *kafka.Config {
+	_, filename, _, _ := runtime.Caller(0)
+	dirname := filepath.Dir(filename)
+	file, err := os.ReadFile(filepath.Join(dirname, "config.yml"))
+	if err != nil {
+		panic(err)
+	}
+
+	cfg := &kafka.Config{}
+	err = yaml.Unmarshal(file, cfg)
+	if err != nil {
+		panic(err)
+	}
+
+	return cfg
+}

internal/consumer.go

@@ -3,14 +3,13 @@ package internal
 import (
 	"context"
 	"errors"
-	"io"
 
 	"github.com/Trendyol/kafka-cronsumer/pkg/kafka"
 	segmentio "github.com/segmentio/kafka-go"
 )
 
 type Consumer interface {
-	ReadMessage() (MessageWrapper, error)
+	ReadMessage(ctx context.Context) (*MessageWrapper, error)
 	Stop()
 }
 
@@ -52,22 +51,21 @@ func newConsumer(kafkaConfig *kafka.Config) *kafkaConsumer {
 	}
 }
 
-func (k kafkaConsumer) ReadMessage() (MessageWrapper, error) {
-	msg, err := k.consumer.ReadMessage(context.Background())
+func (k kafkaConsumer) ReadMessage(ctx context.Context) (*MessageWrapper, error) {
+	msg, err := k.consumer.ReadMessage(ctx)
 	if err != nil {
-		if k.IsReaderHasBeenClosed(err) {
-			return MessageWrapper{}, err
+		if isContextCancelled(err) {
+			k.cfg.Logger.Info("kafka-go context is cancelled")
+			return nil, nil
 		}
-
-		k.cfg.Logger.Errorf("Message not read %v", err)
-		return MessageWrapper{}, err
+		return nil, err
 	}
 
-	return newMessage(msg), err
+	return newMessage(msg), nil
 }
 
-func (k kafkaConsumer) IsReaderHasBeenClosed(err error) bool {
-	return errors.Is(err, io.EOF)
+func isContextCancelled(err error) bool {
+	return errors.Is(err, context.Canceled)
 }
 
 func (k kafkaConsumer) Stop() {

internal/cron.go

@@ -1,6 +1,7 @@
 package internal
 
 import (
+	"context"
 	"time"
 
 	"github.com/Trendyol/kafka-cronsumer/pkg/kafka"
@@ -30,22 +31,12 @@ func (s *cronsumer) WithLogger(logger logger.Interface) {
 }
 
 func (s *cronsumer) Start() {
-	cfg := s.cfg.Consumer
-	_, _ = s.cron.AddFunc(cfg.Cron, func() {
-		s.cfg.Logger.Info("Topic started at time: " + time.Now().String())
-		s.consumer.Start(setConcurrency(cfg.Concurrency))
-		time.AfterFunc(cfg.Duration, s.consumer.Pause)
-	})
+	s.setup()
 	s.cron.Start()
 }
 
 func (s *cronsumer) Run() {
-	cfg := s.cfg.Consumer
-	_, _ = s.cron.AddFunc(cfg.Cron, func() {
-		s.cfg.Logger.Info("Topic started at time: " + time.Now().String())
-		s.consumer.Start(setConcurrency(cfg.Concurrency))
-		time.AfterFunc(cfg.Duration, s.consumer.Pause)
-	})
+	s.setup()
 	s.cron.Run()
 }
 
@@ -54,9 +45,20 @@ func (s *cronsumer) Stop() {
 	s.consumer.Stop()
 }
 
-func setConcurrency(concurrency int) int {
-	if concurrency == 0 {
-		return 1
-	}
-	return concurrency
+func (s *cronsumer) setup() {
+	cfg := s.cfg.Consumer
+
+	s.consumer.SetupConcurrentWorkers(cfg.Concurrency)
+
+	_, _ = s.cron.AddFunc(cfg.Cron, func() {
+		s.cfg.Logger.Info("Topic started at time: " + time.Now().String())
+
+		ctx, cancelFunc := context.WithCancel(context.Background())
+		go s.consumer.Listen(ctx)
+
+		time.AfterFunc(cfg.Duration, func() {
+			s.cfg.Logger.Info("Process Topic PAUSED")
+			cancelFunc()
+		})
+	})
 }

internal/cronsumer.go

@@ -1,14 +1,13 @@
 package internal
 
 import (
+	"context"
 	"time"
 
 	"github.com/Trendyol/kafka-cronsumer/pkg/kafka"
 )
 
 type kafkaCronsumer struct {
-	paused         bool
-	quitChannel    chan bool
 	messageChannel chan MessageWrapper
 
 	kafkaConsumer Consumer
@@ -25,75 +24,49 @@ type kafkaCronsumer struct {
 func newKafkaCronsumer(cfg *kafka.Config, c func(message kafka.Message) error) *kafkaCronsumer {
 	cfg.SetDefaults()
 	cfg.Validate()
+
 	return &kafkaCronsumer{
 		cfg:             cfg,
-		paused:          false,
-		quitChannel:     make(chan bool),
 		messageChannel:  make(chan MessageWrapper),
 		kafkaConsumer:   newConsumer(cfg),
 		kafkaProducer:   newProducer(cfg),
 		consumeFn:       c,
-		maxRetry:        setMaxRetry(cfg.Consumer.MaxRetry),
+		maxRetry:        cfg.Consumer.MaxRetry,
 		deadLetterTopic: cfg.Consumer.DeadLetterTopic,
 	}
 }
 
-func setMaxRetry(maxRetry int) int {
-	if maxRetry == 0 {
-		return 3
-	}
-	return maxRetry
-}
-
-func (k *kafkaCronsumer) Start(concurrency int) {
-	k.Resume()
-	go k.Listen()
-
+func (k *kafkaCronsumer) SetupConcurrentWorkers(concurrency int) {
 	for i := 0; i < concurrency; i++ {
 		go k.processMessage()
 	}
 }
 
-func (k *kafkaCronsumer) Resume() {
-	k.messageChannel = make(chan MessageWrapper)
-	k.paused = false
-	k.quitChannel = make(chan bool)
-}
-
-func (k *kafkaCronsumer) Listen() {
+func (k *kafkaCronsumer) Listen(ctx context.Context) {
 	startTime := time.Now()
 
 	for {
-		select {
-		case <-k.quitChannel:
+		msg, err := k.kafkaConsumer.ReadMessage(ctx)
+		if err != nil {
+			k.cfg.Logger.Errorf("Message could not read, error %v", err)
 			return
-		default:
-			msg, err := k.kafkaConsumer.ReadMessage()
-			if err != nil {
-				continue
-			}
+		}
 
-			if msg.Time.Before(startTime) {
-				k.sendToMessageChannel(msg)
-			} else {
-				k.Pause()
+		if msg == nil {
+			return
+		}
 
-				if err := k.kafkaProducer.Produce(msg, false); err != nil {
-					k.cfg.Logger.Errorf("Error sending next iteration KafkaMessage: %v", err)
-				}
+		if msg.Time.After(startTime) {
+			k.cfg.Logger.Info("Next iteration KafkaMessage has been detected, resending exception topic")
 
-				return
+			if err = k.kafkaProducer.Produce(*msg, false); err != nil {
+				k.cfg.Logger.Errorf("Error sending next iteration KafkaMessage: %v", err)
 			}
+
+			return
 		}
-	}
-}
 
-func (k *kafkaCronsumer) Pause() {
-	if !k.paused {
-		k.cfg.Logger.Info("Process Topic PAUSED")
-		close(k.messageChannel)
-		k.paused = true
-		k.quitChannel <- true
+		k.sendToMessageChannel(*msg)
 	}
 }
 

internal/message.go

@@ -17,8 +17,8 @@ type MessageWrapper struct {
 
 const RetryHeaderKey = "x-retry-count"
 
-func newMessage(msg segmentio.Message) MessageWrapper {
-	return MessageWrapper{
+func newMessage(msg segmentio.Message) *MessageWrapper {
+	return &MessageWrapper{
 		RetryCount: getRetryCount(&msg),
 		Message: kafka.Message{
 			Topic:         msg.Topic,

pkg/kafka/config.go

@@ -55,6 +55,12 @@ type ProducerConfig struct {
 }
 
 func (c *Config) SetDefaults() {
+	if c.Consumer.MaxRetry == 0 {
+		c.Consumer.MaxRetry = 3
+	}
+	if c.Consumer.Concurrency == 0 {
+		c.Consumer.Concurrency = 1
+	}
 	if c.Consumer.MinBytes == 0 {
 		c.Consumer.MinBytes = 10e3
 	}

pkg/kafka/config_test.go

@@ -27,6 +27,8 @@ func TestConfig_SetDefaults(t *testing.T) {
 			name: "should be set to default values when any value is empty",
 			expected: fields{
 				Consumer: ConsumerConfig{
+					MaxRetry:          3,
+					Concurrency:       1,
 					MinBytes:          10e3,
 					MaxBytes:          10e6,
 					MaxWait:           2 * time.Second,