Add support for Kafka's ConsumerRebalanceListener

[pcasaes]

Details #577

Had to create a custom listener interface that also receives the KafkaConsumer beside the assigned topics with partitions. This is necessary for use cases that involves seeking to specific offsets during rebalance/initialization.

[cescoffier]

Amazing! Thanks.

I've made a few comments, but nothing critical.
I'm wondering if this feature does not deserve its own section in the doc. Something explaining how it's used and link to the Kafka documentation. WDYT?

[cescoffier]

Please rebase. I should fix the CI issue.

[pcasaes]

I thought the same thing. I wanted to to get the code out for review first.

[pcasaes]

@cescoffier Added an example in the docs. I thought occurred to me though. We could dispense with the consumer-rebalance-listener.name configuration if we were to use the same channel-name on the listener. What do you think?

Pro: less configuration work.
Con: Multiple inbound connectors won't be able to share a listener.

[cescoffier]

More a question than a comment. I wonder if the listener should return Uni<Void> to indicate when it completes or fails. However, that would not fit very well with the callback registered on the consumer.

At the moment, it requires verifying when the action has completed and I'm not totally sure about failures.

[pcasaes]

I was able to get it working but the solution isn't very elegant. It doesn't solve the fundamental problem which is that we need to wait for the partition assigned Uni to complete before continuing with the poll. I'll take a look at kafka mutiny module and see what can be done.

[pcasaes]

@cescoffier I got it working well enough. The KafkaConsumer class has pause and resume methods which made the work possible.

Did you get chance to think about my comments on #601 (comment) ?

[cescoffier]

@cescoffier Added an example in the docs. I thought occurred to me though. We could dispense with the consumer-rebalance-listener.name configuration if we were to use the same channel-name on the listener. What do you think?

Pro: less configuration work.
Con: Multiple inbound connectors won't be able to share a listener.

@pcasaes Interesting idea. The questions are:

1. do we want to use a default one (if there is one exposed without a name)?
2. do we want to use one having the channel name as name?

About 1:

I'm not sure, it's a fairly advanced use case, you probably don't want to use it unconsciously. So, I would say no: it must have a name.

About 2:

I like the idea. At least the developer is aware of it. It does not reduce the flexibility as you can still have a specific name and configure the attribute in that case. So, you can still have a listener shared by multiple channels.

The question is about channel name or topic name. As it's quite Kafka specific, I would say that it should use the topic name, and not the channel name.

So:

1. if the configuration attribute is not set and there is a listener named after the topic name you are consuming, you use that one. A log message should be added to indicate the choice.
2. if the configuration attribute is set and there is no listener with that name, you fail.
3. if the configuration attribute is set and there is a listener with that name, you use that one.

WDYT?

[pcasaes]

I gave it some thought. I understand why we wouldn't want to use the channel name but the topic name doesn't seem quite right either. You could conceivably set up two connectors against the same topic. How about the consumer group? That makes more sense. It's in the name after all: consumer rebalance listener. And setting up two consumers instances on the same group but with different setups seems incorrect.

EDIT: In a future PR we might want to allow a consumer to read from multiple topics (kafka consumers allow for this) so using the group id will make us future proof.

[pcasaes]

@cescoffier I think it's done. Let me know what you think.

[cescoffier]

The consumer group Id is the right thing to do.

About the pause/resume, so, it should be paused and resumed by the connector. The connector should also handle failures and probably "fails stop" in this case.

[pcasaes]

Went ahead and moved the pause/resume to the connector. In the case of a failure kafka will handle it. If a consumer instance is paused for a certain amount of time it will trigger a consumer rebalance.

[pcasaes]

@cescoffier

1. Used the logger for all logs

2. Removed the merging Multi (that was a relic from an older attempt).

3. We have to resume the consumer for it to trigger a rebalance
   This one was crucial and was discovered by adding a test for this case. So if the rebalance Uni results in an error we wait session.timeout.ms + max.poll.interval.ms before resuming the consumer.

[pcasaes]

I think I can do better with shorter retries.

[pcasaes]

@cescoffier
If assigning fails will keep retrying using an exponential back off until the kafka consumer session can be considered timed out. According to the documentation it's max.poll.interval.ms or max.poll.interval.ms + session.timeout.ms if group.instance.id is not null. Once timed out we'll resume the consumer which will trigger a new consumer re-balance. Added tests to both cases, success on a simple retry and success only after triggering a new consumer re-balance.

[cescoffier]

The logic sounds good. I've made a few cosmetic comments.
I believe one test is missing (when the configured listener cannot be found).

[pcasaes]

All done

[cescoffier]

Amazing! Thanks for the hard work @pcasaes!