feat: add functionality pause and resume for consumer

batch_consumer.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+ "context"
  "time"
 
  "github.com/segmentio/kafka-go"
@@ -17,6 +18,18 @@
  messageGroupLimit int
 }
 
+func (b *batchConsumer) Pause() {
+ b.logger.Info("Batch consumer is paused!")
+ b.pauseConsuming = true
+ b.cancelFn()
+}
+
+func (b *batchConsumer) Resume() {
+ b.logger.Info("Consumer is resumed!")
+ b.pauseConsuming = false
+ b.context, b.cancelFn = context.WithCancel(context.Background())
+}
+
 func newBatchConsumer(cfg *ConsumerConfig) (Consumer, error) {
  consumerBase, err := newBase(cfg, cfg.BatchConfiguration.MessageGroupLimit*cfg.Concurrency)
  if err != nil {

consumer.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+ "context"
  "time"
 
  "github.com/segmentio/kafka-go"
@@ -14,6 +15,18 @@
  consumeFn func(*Message) error
 }
 
+func (c *consumer) Pause() {
+ c.logger.Info("Consumer is paused!")
+ c.pauseConsuming = true
+ c.cancelFn()
+}
+
+func (c *consumer) Resume() {
+ c.logger.Info("Consumer is resumed!")
+ c.pauseConsuming = false
+ c.context, c.cancelFn = context.WithCancel(context.Background())
+}
+
 func newSingleConsumer(cfg *ConsumerConfig) (Consumer, error) {
  consumerBase, err := newBase(cfg, cfg.Concurrency)
  if err != nil {

consumer_base.go

@@ -20,6 +20,12 @@
  // Consume starts consuming
  Consume()
 
+ // Pause pauses consumer, it is stop consuming new messages
+ Pause()
+
+ // Resume resumes consumer, it is start to working
+ Resume()
+
  // WithLogger for injecting custom log implementation
  WithLogger(logger LoggerInterface)
 
@@ -56,6 +62,7 @@
  retryEnabled bool
  transactionalRetry bool
  distributedTracingEnabled bool
+ pauseConsuming bool
 }
 
 func NewConsumer(cfg *ConsumerConfig) (Consumer, error) {
@@ -129,6 +136,10 @@
  close(c.incomingMessageStream)
  return
  default:
+ if c.pauseConsuming {
+ continue
+ }
+
  m := &kafka.Message{}
  err := c.r.FetchMessage(c.context, m)
  if err != nil {
@@ -158,7 +169,8 @@
 }
 
 func (c *base) Stop() error {
- c.logger.Debug("Stop called!")
+ c.logger.Info("Stop called!")
+
  var err error
  c.once.Do(func() {
  c.subprocesses.Stop()

examples/with-pause-resume-consumer/main.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+ "fmt"
+ "github.com/Trendyol/kafka-konsumer/v2"
+ "os"
+ "os/signal"
+ "time"
+)
+
+func main() {
+ consumerCfg := &kafka.ConsumerConfig{
+ Concurrency: 1,
+ Reader: kafka.ReaderConfig{
+ Brokers: []string{"localhost:29092"},
+ Topic: "standart-topic",
+ GroupID: "standart-cg",
+ },
+ RetryEnabled: false,
+ ConsumeFn: consumeFn,
+ }
+
+ consumer, _ := kafka.NewConsumer(consumerCfg)
+ defer consumer.Stop()
+
+ consumer.Consume()
+ fmt.Println("Consumer started...!")
+
+ // You can produce a message via kowl.
+ go func() {
+ time.Sleep(10 * time.Second)
+ consumer.Pause()
+
+ time.Sleep(10 * time.Second)
+ consumer.Resume()
+
+ time.Sleep(10 * time.Second)
+ consumer.Pause()
+
+ time.Sleep(10 * time.Second)
+ consumer.Resume()
+
+ time.Sleep(10 * time.Second)
+ consumer.Pause()
+ }()
+
+ c := make(chan os.Signal, 1)
+ signal.Notify(c, os.Interrupt)
+ <-c
+}
+
+func consumeFn(message *kafka.Message) error {
+ fmt.Printf("Message From %s with value %s \n", message.Topic, string(message.Value))
+ return nil
+}