sceneryback/kafka-best-practices

kafka-best-practices's Issues

sigterm does not stop the consumer

Hi @sceneryback, first of all: thank you very much for these examples. They're very, very useful.

While running it, I was not able to make the consumer.Close() really stop and exit the program (I was not running with go run) while using the StartMultiAsyncConsumer. The program receives the fmt.Println("received signal", <-c), but it does not really stop the consumer.

I've even tried to changed the code to use a sync.WaitGroup to coordinate the goroutines reading from the bufChan to stop, but even with that the consumer.Close hangs forever and it does not stop. Do you have any clue why?

Is mostly like this:

ctx, cancel := context.WithCancel(context.Background())
wg := &sync.WaitGroup{}

consumer, err := StartMultiAsyncConsumer(ctx, wg, cfg)
if err != nil {
panic(err)
}

sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)

select {
case <-ctx.Done():
log.Println("terminating: context cancelled")
case <-sigterm:
log.Println("terminating: via signal")
cancel()
wg.Wait()
consumer.Close() // it hangs forever and never finishes the program
}

I wonder if its really a problem to do not call consumer.Close (I've read the sarama code, but it's unclear if not calling it would lead to message losses or something like that).

why MultiBatch is faster than MultiAsync

I was confused why MultiBatch is faster then MultiAsync.

msgBuf cannot be shared, as ConsumeClaim is being called for diff topics partitions

kafka-best-practices/consumer/consume_multi_batch_handler.go

Line 29 in c72c156

msgBuf batchMessages

since msgBuf is shared. It is going to get data from Topic A and Topic B both
Instead msgBuf should be created separately for each ConsumeClaim