Wait for messages to be processed before committing during shutdown

[wvanbergen]

Previously we would immediately try to commit when finalizing a partition. Now, we wait until we have seen all the offsets we have read from Kafka, up to a given timeout.

@bayan any chance you could try this branch and see if you run into the problem?

This fixes #35. @eapache

[eapache]

LGTM, but that's not a super-strong endorsement.

[wvanbergen]

@bayan You can specify a processing timeout:

config := consumergroup.NewConfig()
config.ChannelBufferSize = 256 // default
config.Offsets.ProcessingTimeout = 60 * time.Second  // default
cg, err := consumergroup.JoinConsumergroup(..., config)
...

If your message processing usually takes long, you have to adjust the 60 second default accordingly. With a channel buffer size of 256 and an average message processing time on 100ms, your ProcessingTimeout should be 25.6 seconds or higher.

[bayan]

Thank you and sorry about the delay.

I originally had some problems where it appeared as though starting one consumer would halt the others, as well as some other weird behaviour.

So I rewrote a simple producer/consumer example today to try to isolate the issue, and it appears to work fine.

@wvanbergen Can you please confirm your formula for setting the ProcessingTimeout?
By your example it looks like: ChannelBufferSize * AverageProcessingTimeInSeconds

The messages I receive are batch file paths, so it takes anywhere from a few seconds and up to several minutes to process each one. I have been setting the ChannelBufferSize to 1 or even 0.

[wvanbergen]

Yup, more precisely it's (ChannelBufferSize + 1) * AverageProcessingTimeInSeconds.

In your case I'd sat ChannelBufferSize (you don't have much performance to gain by buffering), and set the timeout high enough to be able to finish the message that is currently processing, e.g. 10 minutes.

[bayan]

Thanks - that's exactly what I did!