Add test for next generation consumer-group rebalance protocol

[sobychacko]

Starting with 3.7.0, the Kafka client now supports a new consumer group rebalance protocol and categorizes the consumer-group strategy as legacy (classic) vs. consumer. Spring for Apache Kafka supports the new protocol transparently; however, we need to verify this by adding an integration test. See more details here.

[chickenchickenlove]

I am investigating whether I can resolve this issue on my own. However, it seems currently unachievable due to the limitations of KafkaTestKit, which is used to create EmbeddedKafka.

I have created an issue with the apache/Kafka Team to see if there are any other workarounds.
I will let you know if there are any updates. 😅

[artembilan]

@chickenchickenlove ,

Thank you for looking into this!

Would you mind to share more info what the problem have you faced?
What issue have you created on Apache Kafka side?
Did you try to test it against @EmbeddedKafka(kraft = false) which does not use KafkaClusterTestKit?

It is a bit odd to hear that broker behavior has some how an effect on the consumer:

            GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));

            if (groupProtocol == GroupProtocol.CONSUMER)
                return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer);
            else
                return new LegacyKafkaConsumer<>(config, keyDeserializer, valueDeserializer);

[sobychacko]

@chickenchickenlove This is an issue we need to tackle before the GA release, so we were planning to take a look at this, glad that you started on it. Please let us know if disabling kraft as @artembilan suggested works.

[chickenchickenlove]

@sobychacko, @artembilan thanks for you guys comments.
I didn't know about @EmbeddedKafka(kraft = false). i'm going to try to test it ASAP. 👍

This is issue which i reported.
I tested KIP-848 rebalancing with this code. (docker-compose and simple main class). and it did work well.

However, i have problem when i used KafkaTestKit with @EmbeddedKafka.
I think here is the thing.

NOTE: We cannot set controller.quorum.voters the way we want. (always, 0.0.0.0:0)

I may not be very knowledgeable about KRAFT, so this might not be accurate.
It seems that in that state, there is no quorum available for voting.

The AsyncKafkaConsumer sends a HeartBeat to the Broker using the consumer group protocol, but the Broker cannot assign Topic Partitions because there are no quorum voters available to assign.

Do you have any idea how controller.quorum.voters=0.0.0.0:0 works with?

[artembilan]

Thanks for sharing!

So, if we cannot change controller.quorum.voters for that KafkaTestKit, then it sounds like we cannot test this new consumer mode for KRaft environment.
Back in days we also have faced the problem that KafkaTestKit is not able to accept fixed ports for brokers it creates.
Therefore we recommend to use ZK mode kraft = false if you deal with fixed ports.
Might be the case for this new consumer as well: just don't use KRaft!

I wonder if we need to think about changing a default kraft from true to false since it is not able to cover all the use-cases we have around.
At least for time being until Apache Kafka will have those inconveniences fixed...

[sobychacko]

I think staying with kraft = false for this test will help us verify the behavior of this new consumer group faster without worrying about kraft (for the time being, given the issues we ran into).

[chickenchickenlove]

@artembilan, @sobychacko.
Before setting kraft=false in @EmbeddedKafka, I first tested whether KIP-848 works in ZK Mode with docker-compose.
IMHO, we can use new consumer rebalancing protocol in only KRaft mode. 😓😓😓😓

I have created two versions of the docker-compose files.

• ZK mode (link)
• Kraft mode. (link)

After setting up the consumer configurations identically, the consumer sent requests to each broker(ZK, KRaft) and received different results.

In ZK mode, consumer received unsupported API Messages from broker. (link)
On the Otherhand, in KRAFT mode, consumer got metadata of assigned topic and partitions. (link)

Thus, i think we should you KRAFT mode to test new rebalancing protocol.

If we should test it anyway, i have one workaround.
How about using testContainer until testing with EmbeddedKafka is possible? 🤔

• https://mvnrepository.com/artifact/org.testcontainers/kafka
• https://mvnrepository.com/artifact/org.testcontainers/testcontainers

I think it can be one of work around. But of course, we must endure the longer test times and the addition of new dependencies.

What do you think, you guys?

[sobychacko]

@chickenchickenlove Thanks for these findings. Is the fact that the new consumer protocol is not working with ZK a well-known issue? Is there a corresponding Kafka Jira issue that we can link from here? If this is the case, we should mention it in our docs until it is resolved; otherwise, we will see issues related to it coming to this project, thinking that the issue is in Spring for Apache Kafka.

Regarding your question about using test containers, we discussed this as an alternative internally yesterday but opted against bringing it into the project. We don't really need a test for this feature in the framework; we thought that it would be nice to have a test to verify it since this is the first time we are exposing this via Spring for Apache Kafka. The framework didn't have to make any code changes related to this when we upgraded to the 3.7.0 Java client. Maybe you can create a standalone application that uses the new consumer group protocol + testcontainers-based Kafka using KRaft. If anyone wants to see an example of this feature, we can point to this until the test kit issues are resolved. What do you think?

[artembilan]

Or we can opt-in for respective sample where we can use Testcontaiers to demonstrate the feature...