GH-2302: Enable consumer seek only on matching group Id

[bky373]

Backgrounds

• Fixes: consumer group specific offset seeking for AbstractConsumerSeekAware  #2302
• Using @KafkaListener, listeners with different group IDs can exist within the same class.
• Using class which extends AbstractConsumerSeekAware, when seek is performed in one listener, it is executed in all listeners, even if it is not desired.

Changes

• Add a new default method (matchGroupId()) to ConsumerSeekAware. This will allow setting the callback and performing seeks only for the desired group ID. (default: false)
• Override matchGroupId() in AbstractConsumerSeekAware to return true. It would be nice to set this to false so that users should specify the groupId themselves, but in this case the callback from KafkaMessageListenerContainer.ListenerConsumer.setupSeeks() would not be registered, so I set it to true. And this is also to ensure backwards compatibility.

[sobychacko]

@bky373 I see that the general idea is what we discussed. We need to add some documentation, explaining how end users can match the group id.

[sobychacko]

@bky373 I think we may need to take a slightly different approach in this PR. While the solution you have in the PR currently satisfies the immediate use cases at hand, this may not be flexible enough for other wider set of requirements. Take a look at this code snippet you provided in the issue.

@Component
    public static class Listener extends AbstractConsumerSeekAware {

        @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
        public void listen(String payload) {
            System.out.println("Listener received: " + payload);
        }

        @KafkaListener(id = "seekExample3", topics = "seekExample", concurrency = "3")
        public void listen3(String payload) {
            System.out.println("Listener3 received: " + payload);
        }

        public void seekToStart() {
            getSeekCallbacks().forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
        }
    }

While you register the callbacks based on the new method's boolean return value, which works as expected, this is not flexible enough. What if the user wants to seek other listeners based on some other conditions during runtime? The callback registration happens at startup only once, and the callbacks cannot be re-registered during the runtime. Moreover, the seeks must happen on the listener threads, which the PR is doing currently. A better solution is a hybrid of what we discussed in the issue.

After the getSeekCallbacks(), the user needs to do some filtering operations to get the group IDs. Currently, the ConsumerSeekCallback returned doesn't give you the group ID. This is where you need to make the core changes so that the ConsumerSeekCallback adds a new public API for retrieving the group ID. Could you rework the PR to accommodate this new design? Thanks!

CC @artembilan for any additional comments.

[bky373]

Okay, I see. I guess we'll need a flexible implementation.
I'll have to think about this a bit more and work on it
Thank you so much for working with me on this.

[bky373]

@sobychacko
a81a06d (It is better to see Files changed)
I have added a groupId getter to ConsumerSeekCallback. This approach also minimizes changes, and based on the new tests created, users can utilize it as needed in their preferred manner. Your feedback on this would be greatly appreciated. Thank you.

[sobychacko]

@bky373 I took a look at your latest changes. It looks good to me. The changes look very simple, and it also makes the API flexible. I also get the other issue you have encountered, and it is a good idea to work on it in a separate PR. I added some review comments in the new test class.

[bky373]

@bky373 I see that the general idea is what we discussed. We need to add some documentation, explaining how end users can match the group id.

@sobychacko The implementation has changed, but should we add documentation on how to enable seek for only certain groupIds in seek.adoc?

[sobychacko]

Yes, I think we need to add this in the seek section of the ref docs. Also, since we are adding a new public API, we need to mention this in the whats-new section.

[bky373]

Yes, I think we need to add this in the seek section of the ref docs. Also, since we are adding a new public API, we need to mention this in the whats-new section.

@sobychacko
Please check the documentation.

[artembilan]

Please, find some nit-picks from me.
In general it looks good and very close to merge.

Thank you!