Supports multi-topic in CommittableOffsetBatch

[geirolz]

The purpose of this PR is to resolve issue #347 at its root allowing the CommittableOffsetBatch to process records from different topics.
It's very useful when you consume from multiple topics merging the streams.

I didn't address the tests for now, I'll do asap, I the meanwhile let me know what you think about the solution.

Open points:

• Not sure about the error message/exception class name/message in case we can't retrieve the commit action for a specific topic. Any suggestions?

[geirolz]

Hi @bplommer did you have the chance to take a look at this?

[aartigao]

Hi @geirolz!

I've read this PR and the issue #347 and I'm confused about the usage of Topics and Consumer Groups. I'm trying to understand why current CommittableOffsetBatch implementation is not sufficient and what's the use case scenario for that.

For sure I'm missing something (and I'd assume that the issue comes when using multiple Consumer Groups, not Topics, correct me if I'm wrong). If ConsumerSettings only allows a single Group ID how come it ends a stream with multiple Consumer Groups? Is it because you merge multiple KafkaConsumer.records streams?

If that's the case, I'd say that this is something to be handled in user land 🤔 Because by reading the Scaladoc of CommitableOffsetBatch#consumerGroupIds:

There might be more than one consumer group ID in the set
if offsets from multiple consumers, with different group
IDs, have accidentally been mixed. The set might also be
empty if no consumer group IDs have been specified.

The key point is: accidentally been mixed. That suggests it's forbidden, and I think it's right because every KafkaConsumer has its own underlying session/connection with Kafka, and manage its offsets isolated from other groups.

[geirolz]

Hi @aartigao, I've noticed this problem in my previous company. We where consuming from multiple topics merging all messages into one single stream consumed by and internal processor, once processed we had to commit and here the problem came out.

Moreover as #347 this behaviour it's a bit implicit by semantic.

I don't see the disadvantages of supporting this but I may missed something since it was quite easy to implement. I wonder what's the reason behind this design decision then

[aartigao]

Thanks for sharing the graph, now to me it makes sense what's happening.

Most probably those consumers have a different Consumer Group, right?

I'd need to dig deep on this when I have time but I'm afraid that depending on how the underlying Kafka Client works it may not be possible to fix this at lib level. That would explain the decision to support a single Consumer Group only.

Also, just to give more context, a Consumer Group can manage multiple Topics offsets (in fact, multiple TopicPartition). Maybe the case in your company's design was to have a 1:1 relation between Topic and Consumer Group, but that's not the general case. This is where I was confused about before, by looking at the PR implementation using the Topic name as the Map's key.

But maybe I'm wrong, ofc 😅 I'll try to clarify the situation.

In the meantime, thanks for giving your time to create and clarify this PR 🙌🏼