KAFKA-18529: ConsumerRebootstrapTest should run for async consumer #18554

TaiJuWu · 2025-01-15T14:31:19Z

Jira: https://issues.apache.org/jira/browse/KAFKA-18529

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

lianetm · 2025-01-15T18:14:14Z

Thanks for the patch @TaiJuWu ! Could you please merge trunk latest changes? Thanks!

TaiJuWu · 2025-01-15T18:43:00Z

Thanks for the patch @TaiJuWu ! Could you please merge trunk latest changes? Thanks!

Done~~~~~~

lianetm · 2025-01-15T19:09:46Z

core/src/test/scala/integration/kafka/api/ConsumerRebootstrapTest.scala

-    assertEquals(1, getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly.count())
-    val args = getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
+    assertEquals(2, getTestQuorumAndGroupProtocolParametersAll.count())
+    val args = getTestQuorumAndGroupProtocolParametersAll
      .findFirst().get.get


if we're only keeping the first entry from the full protocol list we still won't be testing with both consumers right?

Thanks for catching it, update!

m1a2st

Hello @TaiJuWu, Thanks for this PR, left a comment

m1a2st · 2025-01-16T12:01:45Z

core/src/test/scala/integration/kafka/api/ConsumerRebootstrapTest.scala

-      Arguments.of((args :+ true):_*),
-      Arguments.of((args :+ false):_*)
-    )
+    assertEquals(2, getTestQuorumAndGroupProtocolParametersAll.count())


Should we still need this assertion?

makes sense to me to remove it, this test shouldn't care about the elements in the list, just make sure we use them all right? (this is the only comment left here imo @TaiJuWu, please take a look when you can and I'll take another look too). Thanks!

Thanks, remove it.

lianetm

Thanks! LGTM

lianetm · 2025-01-17T16:04:41Z

Hey @TaiJuWu, could you please merge trunk latest changes? (some changes just went in on these files). Thanks!

TaiJuWu · 2025-01-17T18:55:23Z

Hey @TaiJuWu, could you please merge trunk latest changes? (some changes just went in on these files). Thanks!

Done 😀

kirktrue

I'm curious about these changes. Do all of these tests actually create a Consumer? I don't see that immediately when looking at the test code 🤔

lianetm · 2025-01-17T22:21:57Z

They do by using the createConsumer and directly new KafkaConsumer (looking for those will show the usages).

That being said, good point that in the admin there seems to be some using only admin/producer, so we could review and clean up a bit if the consumer grp param if not needed. Thanks for pointing it out @kirktrue! Could you take a look @TaiJuWu ? Thanks!

kirktrue

Thanks for the PR @TaiJuWu!

I'm not sure if I'm confused about the intent here. The majority of the integration tests that were updated to include the group protocol don't actually create any Consumer instances during the test. With this change, there are tens of tests that run twice unnecessarily. Is there a reason that I'm not seeing that these should all be changed?

Thanks!

kirktrue · 2025-01-17T22:39:32Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testDescribeConfigWithOptionTimeoutMs(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:40:06Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testCreatePartitionWithOptionRetryOnQuotaViolation(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:40:25Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testDescribeUserScramCredentials(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:40:37Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testDescribeTransactions(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:40:52Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testDescribeTransactionsTimeout(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:53:28Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testIncrementalAlterConfigsForLog4jLogLevelsDoesNotWorkWithInvalidConfigs(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:53:40Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:53:49Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:53:55Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testListClientMetricsResources(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T22:54:05Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testCreateTopicsReturnsConfigs(quorum: String, groupProtocol: String): Unit = {


This test does not create any Consumers, so this change should be reverted.

kirktrue · 2025-01-17T23:03:30Z

To test whether or not a test needed to be updated, I changed the code that creates a Consumer:

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
index 74592972b9..fe70aebcee 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
@@ -56,6 +56,9 @@ public class ConsumerDelegateCreator {
     public <K, V> ConsumerDelegate<K, V> create(ConsumerConfig config,
                                                 Deserializer<K> keyDeserializer,
                                                 Deserializer<V> valueDeserializer) {
+        if (true)
+            throw new RuntimeException("Intentionally blocking attempt to create a Consumer!");
+
         try {
             GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));
 
@@ -79,6 +82,9 @@ public class ConsumerDelegateCreator {
                                                 SubscriptionState subscriptions,
                                                 ConsumerMetadata metadata,
                                                 List<ConsumerPartitionAssignor> assignors) {
+        if (true)
+            throw new RuntimeException("Intentionally blocking attempt to create a Consumer!");
+
         try {
             GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));

Then I ran PlaintextAdminIntegrationTest locally.

Are there other changes in the queue that are going to start creating Consumers in these tests?

chia7712 · 2025-01-18T02:08:59Z

I agree with @kirktrue's suggestion that some tests do not utilize a consumer. Therefore, running these tests twice is redundant.

TaiJuWu · 2025-01-18T04:07:55Z

Thanks @kirktrue for good point and suggestion , I will revisit all tests.

TaiJuWu · 2025-01-18T05:35:29Z

Revert some change and do follow check:

1. Apply @kirktrue patch 
2. check all failure use two protocol but there are three exceptions
3. Successful case only run one time

three exception is

1. testDeleteRecordsAfterCorruptRecords, this like behavior difference but I have not checked totally.
2. testDescribeClassicGroups, this is for classic protocol
3. testListGroups, already include both protocol

Also apply https://github.com/apache/kafka/pull/18554/files#r1920843220

lianetm · 2025-01-21T17:55:49Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

  @Timeout(10)
-  def testDescribeUserScramCredentialsTimeout(quorum: String, groupProtocol: String): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("kraft"))


agree with this change, but interesting, this shows that PlainTestAdminIntegration was already running for async, even in cases where it was not needed, right?

So checking all these admin tests that were running for both consumers, I notice that testLogStartOffsetCheckpoint is somehow similar to this one, should we apply the same fix to it?

It's running for both but it really only uses a consumer to subscribeAndWaitForAssignment, which is already covered in many consumer-specific tests, is it valuable in this testLogStartOffsetCheckpoint? (seems unrelated). What do you think @kirktrue ? I see you enabled it for both consumers as part of #17670.

Also should we rename the jira/PR? It's only the ConsumerRebotstrap that was not running for async

Yes, at least one other committer has pushed back on enabling both consumer implementations anywhere that a consumer is created. I filed KAFKA-18087 to remedy that, but I haven't yet thought about the heuristic to apply to know when it's "safe" to just test the one consumer or the other.

To address your fourth question, yes, ideally we should update the Jira and PR to accurately reflect what's changing.

Rename jira and PR title.

but I haven't yet thought about the heuristic to apply to know when it's "safe" to just test the one consumer or the other.

we are on the same page :)

It's difficult to definitively confirm that "this functionality of the consumer is already covered by another test." Therefore, a simpler approach is to test both consumers if the test requires the use of a consumer.

Maybe we can revisit only the "slow" tests?

agree, that could be a safe start.

Also, it could make sense to start by reviewing Admin and Producer tests that are running for both consumers. It's probably on those where we can have some good candidates to rollback.

lianetm · 2025-01-23T16:54:46Z

@TaiJuWu could you merge trunk latest changes to get rid of the test failures we have here? (ex. fixed in #18612). Thanks!

TaiJuWu · 2025-01-24T05:10:38Z

@TaiJuWu could you merge trunk latest changes to get rid of the test failures we have here? (ex. fixed in #18612). Thanks!

Done :)

…pache#18554) Reviewers: Kirk True <[email protected]>, Chia-Ping Tsai <[email protected]>, Lianet Magrans <[email protected]>

TaiJuWu added 2 commits January 15, 2025 13:51

enable consumerRebootstrapTest for asunC

3963703

conver plaintextConsumer to async

123b330

github-actions bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) labels Jan 15, 2025

chia7712 added the ci-approved label Jan 15, 2025

Merge branch 'trunk' into KAFKA-18529-ConsumerRebootstrapTest

80c973a

lianetm reviewed Jan 15, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Jan 16, 2025

fix missed run consumer

e906ef1

m1a2st reviewed Jan 16, 2025

View reviewed changes

address Ken's comments

8e803ea

lianetm approved these changes Jan 16, 2025

View reviewed changes

m1a2st approved these changes Jan 17, 2025

View reviewed changes

Merge branch 'trunk' into KAFKA-18529-ConsumerRebootstrapTest

8517c42

kirktrue reviewed Jan 17, 2025

View reviewed changes

kirktrue added consumer KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) clients labels Jan 17, 2025

kirktrue suggested changes Jan 17, 2025

View reviewed changes

revisit all plaintextConsumer

7af5f1d

revert two case

560df51

github-actions bot added the small Small PRs label Jan 18, 2025

TaiJuWu force-pushed the KAFKA-18529-ConsumerRebootstrapTest branch from 226a078 to 560df51 Compare January 18, 2025 05:43

address Kirk's comments

7403abb

lianetm reviewed Jan 21, 2025

View reviewed changes

TaiJuWu changed the title ~~KAFKA-18529: PlaintextAdminIntegrationTest and ConsumerRebootstrapTest should run for async consumer~~ KAFKA-18529: ConsumerRebootstrapTest should run for async consumer Jan 22, 2025

Merge branch 'trunk' into KAFKA-18529-ConsumerRebootstrapTest

48e482c

lianetm merged commit 023f9c2 into apache:trunk Jan 24, 2025
9 checks passed

TaiJuWu deleted the KAFKA-18529-ConsumerRebootstrapTest branch January 25, 2025 04:33

		@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
		@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))

KAFKA-18529: ConsumerRebootstrapTest should run for async consumer #18554

KAFKA-18529: ConsumerRebootstrapTest should run for async consumer #18554

Uh oh!

Conversation

TaiJuWu commented Jan 15, 2025

Committer Checklist (excluded from commit message)

Uh oh!

lianetm commented Jan 15, 2025

Uh oh!

TaiJuWu commented Jan 15, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

m1a2st left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm Jan 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

lianetm commented Jan 17, 2025

Uh oh!

TaiJuWu commented Jan 17, 2025

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

lianetm commented Jan 17, 2025

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kirktrue commented Jan 17, 2025

Uh oh!

chia7712 commented Jan 18, 2025

Uh oh!

TaiJuWu commented Jan 18, 2025

Uh oh!

TaiJuWu commented Jan 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lianetm Jan 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm Jan 16, 2025 •

edited

Loading

TaiJuWu commented Jan 18, 2025 •

edited

Loading

lianetm Jan 21, 2025 •

edited

Loading

lianetm Jan 23, 2025 •

edited

Loading