Flow control support

[martindam]

This is more a feature request and an issue: While working with this project and the spring-integration-kafka project, I faced the problem that KafkaConsumer can fetch messages much faster than I can consume them. With the current implementation the batch of messages are read and each one is pushed downstream for processing. However, if it takes more than session.timeout.ms to process the batch (which can be thousands of messages if there is a large kafka lag) a rebalance will be triggered as KafkaConsumer was not able to send heart beats (it happens while calling poll()). This could be solved by placing a queueChannel after the KafkaMessageListenerContainer, but that just moves the problem as at some point that queue is going to fill up or I run out of memory.

So my suggestion is this: Add support for some flow control in the KafkaMessageListenerContainer so I can start/pause the KafkaConsumer when my downstream processing pipe is backed up to prevent filling up all buffers. KafkaConsumer supports start/pause out of the box, so it allows to pause the consumer while it keeps sending heart beats (with poll()) but not receive more messages until it can be consumed downstream. This could then be wrapped in a integration component that could manage the flow control based on MessageDeliveryExceptions or similar.

I did an implementation like the above, but it required to copy the existing KafkaMessageListenerContainer to make it work due to most of the logic is in the inner class.

[garyrussell]

Good suggestion; thanks - feel free to issue a pull request - be sure to follow the contribution guidelines - especially "signing" the CLA. We need to release the GA by the end of the month so that means we need to get the (feature complete) release candidate out in the next day or so.

If you don't think you could turn around a PR in that timeframe, please share your code and one of us can take a stab at it (but we'd still need you to "sign" the CLA - it currently doesn't have Spring Kafka in the dropdown, you can choose Spring Integration instead).

Thanks!!

[garyrussell]

I think we should consider this a bug.

[martindam]

Will submit my code as it is and you can have a look at it. I should be able to make the needed change within a day.
I already signed the CLA for spring-integration-kafka earlier so that should be okay right? Also, do you have a eclipse style format you typically use?

[martindam]

Have a look at #85

[garyrussell]

@martindam Here is my approach: #91

I believe it satisfies both use cases (pause for slow listener, pause if the listener throws a specific exception).