Mismatch Between Registered and Found Callbacks in Multi-group Listener Cases

[bky373]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

For example:

3.2

Describe the bug

• When using AbstractConsumerSeekAware in a multi-group listener scenario, there are cases where the number of registered callbacks differs from the number of discovered callbacks. Let's consider the examples below.

• Case 1:

  • Topic partitions: 1
  • Consumer groups: 2 (group1, group2)
  • Registered callbacks: 2 (group1 thread callback and group2 thread callback)
  • Discovered callbacks: 1 (either group1 thread callback or group2 thread callback — only one callback exists assigned to topic partition 0)

• Case 2 - with concurrency:

  • Topic partitions: 2
  • Consumer groups: 2 (group1, group2)
  • Concurrency: 2 per group
  • Registered callbacks: 4
    • group1-0 thread callback, group1-1 thread callback, group2-0 thread callback, group2-1 thread callback
  • Discovered callbacks: 2
    • 1 thread assigned to partition 0 and 1 thread assigned to partition 1 (uncertain whether they belong to group1 or group2 threads)

Expected behavior

• The number of discovered callbacks should be equal to the number of registered callbacks.

Sample

• 009d8b9

[bky373]

I've created a V2 suggestion for AbstractConsumerSeekAware in here. However, I also think that, as you mentioned before, not allowing multi-group listeners (fast-fail approach) could be beneficial in avoiding confusion. Of course, enforcing separation could potentially impact code flexibility. Please feel free to share your thoughts.

[bky373]

@sobychacko Hi, Is it okay to open PR for this later?

[artembilan]

No problem, @bky373 !

We had a chat with @sobychacko yesterday about this matter and we find it as a bit awkward target application design when you have several @KafkaListener methods in the same service for the same group.
Or even more, if you have several @KafkaListeners with different groups but same topic.
All those are going to create their own KafkaConsumer and be registered this or other way in Kafka group management.
This cause some resources (and network) overhead.

In case of same group only one of those listener methods is always going to be active since consumer group management.
So, from running application perspective some of your code is always dead, but those resources are still engaged: connection, polling cycle, group coordinator etc.

In case of multi-group but same topic all of those @KafkaListener methods are going to be called.
However here is a catch: all of them are going to receive same record from the partition.
So, what is the point in multi-group in same service since you always can receive that record and do with it as many logic as you need.
Why make a a heavy multi-consumer for the same record if you simple can multiply it in a single @KafkaListener?

We understand that this is something no prevented by Kafka client itself, but we always have to be mindful on engineering phase to not overhead all the mentioned resource when we can do the same just in-memory.

I had an idea to fail fast when we encounter such a situation (which indeed leads to the problem described by this issue), however we ended up to document it some way to discourage a configuration I've just explained.
It's like SQL injection via string concatenation: it is a problem, but not of the library, rather target application and its developers.

Sorry for side-tracking, but we agreed that fix for this issue has to in.
So, take your time and we will be happy to review it and merge!

[bky373]

@artembilan
Thank you for digging deeper!

Yes, we need to look at both of the cases you mentioned.

• Case 1 - In case of same group
  • This is where the application developer needs to be careful, as you said. If it's the same group looking at the same topic, it is recommended to adjust the concurrency rather than writing multiple @KafkaListeners.
• Case 2 - In case of multi-group but same topic
  • This link shows an example most similar to what I had in mind. I know it's not much, but I think this requirement exists sometimes.

    @KafkaListener(topics = "books", groupId = "books-content-search")
    public void bookContentSearchConsumer(BookEvent event) {
        logger.info("Books event received for full-text search indexing => {}", event);
    }
    
    @KafkaListener(topics = "books", groupId = "books-price-index")
    public void bookPriceIndexerConsumer(BookEvent event) {
       logger.info("Books event received for price indexing => {}", event);
    }

  • Also, the use of multi-group listeners has already been suggested in retry.
    • https://docs.spring.io/spring-kafka/reference/retrytopic/multi-retry.html

These two examples in Case 2 make me think one more time.
Maybe instead of preventing multi-listeners using @KafkaListeners, it would be beneficial for users to have more flexibility and versatility in providing functionality for them.

Thanks again for your insightful look at this issue. I'll open a PR on this soon.

[bky373]

@artembilan Hi, I have a minor question. From which version will this be available? I know you are preparing for 3.3.0-M1. Will it apply to that?