client/Producer: Close() is not flushing current inflight messages

[brenol]

Hi, found a issue with the library using (*Producer).Close() while messages are still inflight. It does not seem to be flushing them.

Using (*Environment).Close() works, though.

Thanks!

[Gsantomaggio]

Hi @brenol ,
Thank you for reporting. We will have a look!

[brenol]

@Gsantomaggio sorry, I forgot to add sample code. Here it is: https://play.golang.org/p/FZzY5maPkWL

Sample output, using env.Close():

$ go run main.go
2021/11/08 14:50:54 produced 67303 messages
2021/11/08 14:50:55 got close notification. Produced: 67303, Consumed: 67303

Using prod.Close():

go run main.go
2021/11/08 14:51:17 produced 65994 messages
2021/11/08 14:51:18 got close notification. Produced: 65994, Consumed: 56417

p.s. thank you for the great work!

[Gsantomaggio]

Ok, it can be improved, but you have the messages on failed status.

see:
https://play.golang.org/p/pZUN7tN8Ss4

you will have:

2021/11/08 19:35:31 produced 46295 messages
2021/11/08 19:35:34 got close notification. Produced: 46295, Consumed: 46200 , failed 95 - Total 46295 

[brenol]

Makes sense - I noticed there were missing messages when I used the PublishConfirm example from the docs.

Just reported because it took me sometime to notice my code was wrong and not all messages were being flushed. Wanted to do some graceful shutdown on my application.

Again, thank you for the great work! 🎈

[Gsantomaggio]

The producer.send() is asynchronous and stopping while publishing can raise this kind of problem.

A possible workaround is to put an sleep before close so there is time to append and send all the pending messages:

time.Sleep(1 * time.Second)
if err := prod.Close(); err != nil {
    log.Println("could not close producer:", err)
}

I will investigate btw how to improve it.

[brenol]

Thanks! I ended up injecting env.Close though. Do you think it's better to sleep? (perhaps using 2 * producer.GetOptions().BatchPublishingDelay).

[Gsantomaggio]

Do you think it's better to sleep? (perhaps using 2 * producer.GetOptions().BatchPublishingDelay).

I mean it is ugly but it works :)

This fix will help that basically it waits for BatchPublishingDelay