feat: stop and start consumers

[mhmtszr]

Introduction
We may need to turn our consumers on and off while the application is running, depending on our needs. Currently, it is not possible to continue consuming again after consumer is closed.

[oguzhaneren]

There is already a feature in kafka clients called pause/resume consumer. it will be good implement this feature in consumer mechanism. there are some good benefits except stopping/start consumer:

• if consumer pauses kafka client sends heartbeats in background thread so kafka will not enter the rebalance status so this will help to process long-running kind of messages. if your handling thread is busy more than 1 second we can call consumer.pause to eliminate rebalancing.
• another great feature can be implementing "control-flow" or "backpressure" depending on processing time or buffer limit. for example: if client consume 10K message or processing consumed message time greater than X time you can pause to consumer until all received messages processed.
• we can implement delaying messages using thread block without thinking about rebalancing, because when in pause status kafka client will continue to send heartbeat in background thread. if we implement this kind of delaying mechanism this will support persistent retry mechanism with configurable delay time for each message using kafka headers.

Resources:

• https://www.confluent.io/blog/kafka-consumer-multi-threaded-messaging/
• https://medium.com/naukri-engineering/retry-mechanism-and-delay-queues-in-apache-kafka-528a6524f722

[Abdulsametileri]

Currently, you can workaround this such as

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)
	consumer.Consume()

	go func() {
		fmt.Println("closing consumer")
		consumer.Stop()

		time.Sleep(10 * time.Second)

		newConsumer, _ := kafka.NewConsumer(consumerCfg) // take a fresh instance

		fmt.Println("start new fresh consumer")
		newConsumer.Consume()
	}()

	fmt.Println("Consumer started...!")

	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", message.Topic, string(message.Value))
	return nil
}

[Abdulsametileri]

cron.AddFunc(config.StartingTime, func() {
		consumer, err := kafka.NewConsumer(config)
		if err != nil {
			panic("error initializing kafka consumer " + err.Error())
		}

		consumer.Consume()

		time.AfterFunc(config.WorkingDuration, func() {
			consumer.Stop()
		})
	})

[Abdulsametileri]

Implemented in v2.1.9; thank you for all your support <3