KAFKA-15544: Enable integration tests for new consumer #14758

AndrewJSchofield · 2023-11-14T17:02:09Z

This PR parameterizes the consumer integration tests so they can be run against
the existing "generic" group protocol and the new "consumer" group protocol
introduced in KIP-848.

The KIP-848 client code is under construction so some of the tests do not run on
both variants to start with, but the idea is that the tests can be enabled as the gaps
in functionality are closed.

Committer Checklist (excluded from commit message)

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

Merge commits from trunk

kirktrue

Thanks for the PR @AndrewJSchofield!

kirktrue · 2023-11-14T21:07:01Z

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

-    def assertNoMetric(broker: KafkaServer, name: String, quotaType: QuotaType, clientId: String): Unit = {
-        val metricName = broker.metrics.metricName("throttle-time",
-                                  quotaType.toString,
-                                  "",
-                                  "user", "",
-                                  "client-id", clientId)
-        assertNull(broker.metrics.metric(metricName), "Metric should not have been created " + metricName)
+    def assertNoMetric(broker: KafkaBroker, name: String, quotaType: QuotaType, clientId: String): Unit = {
+      val metricName = broker.metrics.metricName("throttle-time",
+        quotaType.toString,
+        "",
+        "user", "",
+        "client-id", clientId)
+      assertNull(broker.metrics.metric(metricName), "Metric should not have been created " + metricName)
    }
-    servers.foreach(assertNoMetric(_, "byte-rate", QuotaType.Produce, producerClientId))
-    servers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Produce, producerClientId))
-    servers.foreach(assertNoMetric(_, "byte-rate", QuotaType.Fetch, consumerClientId))
-    servers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Fetch, consumerClientId))
+    brokers.foreach(assertNoMetric(_, "byte-rate", QuotaType.Produce, producerClientId))
+    brokers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Produce, producerClientId))
+    brokers.foreach(assertNoMetric(_, "byte-rate", QuotaType.Fetch, consumerClientId))
+    brokers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Fetch, consumerClientId))

-    servers.foreach(assertNoMetric(_, "request-time", QuotaType.Request, producerClientId))
-    servers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Request, producerClientId))
-    servers.foreach(assertNoMetric(_, "request-time", QuotaType.Request, consumerClientId))
-    servers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Request, consumerClientId))
+    brokers.foreach(assertNoMetric(_, "request-time", QuotaType.Request, producerClientId))
+    brokers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Request, producerClientId))
+    brokers.foreach(assertNoMetric(_, "request-time", QuotaType.Request, consumerClientId))
+    brokers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Request, consumerClientId))


So this is incorrect in trunk, right?

In this context, servers and brokers are interchangeable. This change makes the test work for ZK or KRaft. Previously, it was ZK-only.

kirktrue · 2023-11-14T21:13:04Z

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

@@ -66,6 +66,7 @@ abstract class AbstractConsumerTest extends BaseRequestTest {


  override protected def brokerPropertyOverrides(properties: Properties): Unit = {
+    properties.setProperty(KafkaConfig.NewGroupCoordinatorEnableProp, "true") // enable KIP-848 group coordinator


At some point we need to test that if the broker's configuration has this set to false that the client won't up and explode. That's a future task...

kirktrue · 2023-11-14T21:17:39Z

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

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package kafka.api
-
-import kafka.utils.TestUtils.waitUntilTrue
-import org.junit.jupiter.api.Assertions.{assertNotNull, assertNull, assertTrue}
-import org.junit.jupiter.api.Test
-
-import java.time.Duration
-import scala.jdk.CollectionConverters._
-
-class BaseAsyncConsumerTest extends AbstractConsumerTest {
-  val defaultBlockingAPITimeoutMs = 1000
-
-  @Test
-  def testCommitAsync(): Unit = {
-    val consumer = createAsyncConsumer()
-    val producer = createProducer()
-    val numRecords = 10000
-    val startingTimestamp = System.currentTimeMillis()
-    val cb = new CountConsumerCommitCallback
-    sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)
-    consumer.assign(List(tp).asJava)
-    consumer.commitAsync(cb)
-    waitUntilTrue(() => {
-      cb.successCount == 1
-    }, "wait until commit is completed successfully", defaultBlockingAPITimeoutMs)
-    val committedOffset = consumer.committed(Set(tp).asJava, Duration.ofMillis(defaultBlockingAPITimeoutMs))
-    assertNotNull(committedOffset)
-    // No valid fetch position due to the absence of consumer.poll; and therefore no offset was committed to
-    // tp. The committed offset should be null. This is intentional.
-    assertNull(committedOffset.get(tp))
-    assertTrue(consumer.assignment.contains(tp))
-  }
-
-  @Test
-  def testCommitSync(): Unit = {
-    val consumer = createAsyncConsumer()
-    val producer = createProducer()
-    val numRecords = 10000
-    val startingTimestamp = System.currentTimeMillis()
-    sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)
-    consumer.assign(List(tp).asJava)
-    consumer.commitSync()
-    val committedOffset = consumer.committed(Set(tp).asJava, Duration.ofMillis(defaultBlockingAPITimeoutMs))
-    assertNotNull(committedOffset)
-    // No valid fetch position due to the absence of consumer.poll; and therefore no offset was committed to
-    // tp. The committed offset should be null. This is intentional.
-    assertNull(committedOffset.get(tp))
-    assertTrue(consumer.assignment.contains(tp))
-  }
-}


Good riddance!

kirktrue · 2023-11-14T21:18:25Z

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

-  @Test
-  def testSimpleConsumption(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic", "consumer"))


Is it possible to have a different type of 'source' that can be defined once vs. on each test? More of a nit, but curious.

Perhaps, but I want to be able to have a different array for each test to enable them to be turned on individually.

This has been replaced with a MethodSource that is capable of returning whatever combination we want.

kirktrue · 2023-11-14T21:20:45Z

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

-    val partition = 0;
+    val partition = 0


This ain't Java, come on!

kirktrue · 2023-11-14T21:21:15Z

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

-    TestUtils.waitUntilTrue(() => {
-      this.zkClient.topicExists(topic)
-    }, "Failed to create topic")
-


Why don't we need this now?

I'm not convinced we ever did. Again, this is ZK-specific. The tests work on both variants without this check.

I agree with @AndrewJSchofield . Method createTopics() already verifies that the topic was created.

kirktrue · 2023-11-15T00:18:17Z

@philipnee can you tag with ctr and KIP-848 🥺

AndrewJSchofield · 2023-11-21T09:06:53Z

Converting back to draft. This PR depends on #14801 being merged. This is ready for review, but will not build cleanly until the other PR lands.

kirktrue · 2023-11-21T17:50:54Z

@cadonna / @lucasbru—can we add the ctr label to this PR?

cadonna · 2023-11-21T17:55:29Z

Done!

cadonna

Thanks for the PR, @AndrewJSchofield !

Here my comments.

cadonna · 2023-11-22T09:36:52Z

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

-  @Test
-  def testAutoCommitOnCloseAfterWakeup(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic")) // close() is not committing offsets in consumer group protocol


This is just a transient state, isn't it? At least https://issues.apache.org/jira/browse/KAFKA-15327 says that committing on close is planned. If KAFKA-15327 is still valid, can we formulate this comment accordingly like close() is not committing offsets in consumer group protocol for now but it should when implementation is complete.

I agree with the idea, but I will use a less wordy formulation such as "temporarily". I expect a bunch of these will be resolved in the next two weeks and we can run most of the tests across both consumers.

cadonna · 2023-11-22T09:46:58Z

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

-  @Test
-  def testPartitionsForInvalidTopic(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic")) // partitionsFor not supported for consumer group protocol


See my comment above about the functionality being supported in future.

cadonna · 2023-11-22T09:53:58Z

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

-  @Test
-  def testMultiConsumerSessionTimeoutOnStopPolling(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic")) // ConsumerRebalanceListener not supported for consumer group protocol


I guess the ConsumerRebalanceListener is also something that will be supported in the future by the consumer group protocol.

cadonna · 2023-11-22T09:54:17Z

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

-  @Test
-  def testMultiConsumerSessionTimeoutOnClose(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic")) // ConsumerRebalanceListener not supported for consumer group protocol


cadonna · 2023-11-22T09:55:06Z

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

-  @Test
-  def testInterceptors(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic")) // Consumer interceptors not implemented for consumer group protocol


Will also interceptors be supported?

cadonna · 2023-11-22T10:39:42Z

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

+    // No valid fetch position due to the absence of consumer.poll; and therefore no offset was committed to
+    // tp. The committed offset should be null. This is intentional.


Could you make this clear in the name of the test instead of using a inline comment?

cadonna · 2023-11-22T10:39:56Z

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

+    // No valid fetch position due to the absence of consumer.poll; and therefore no offset was committed to
+    // tp. The committed offset should be null. This is intentional.


See my comment above

cadonna · 2023-11-22T10:42:55Z

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

+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = Array("generic")) // partitionsFor not implemented in consumer group protocol


See my other comments about functionality that will be supported in the future.

cadonna · 2023-11-22T10:56:07Z

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

-  def testConsumingWithEmptyGroupId(): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("consumer")) // Null group ID not supported with consumer group protocol
+  def testNullGroupIdNotSupported(groupProtocol: String): Unit = {


This name suggests that a group ID of null is not supported in general with manual assignment. However, null is supported as group ID with manual assignment. Just commits are not supported with group ID = null. Could you please use a more descriptive name for this test like testNullGroupIdNotSupportedIfCommitting?

cadonna · 2023-11-22T10:59:07Z

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

+
+    // Check committed offsets twice with same consumer
+    assertEquals(numRecords, consumer.committed(Set(tp).asJava).get(tp).offset)
+    assertEquals(numRecords, consumer.committed(Set(tp).asJava).get(tp).offset)


Wouldn't it be better to consumer and commit some records before retrieving the committed offsets a second time?

I didn't write this test and there's no comment by the author. I suppose this was done specifically to catch a problem and I prefer not to mess around with it when I don't understand.

dajac · 2023-11-22T11:02:19Z

@AndrewJSchofield We are about to merge #14781. Is it possible to build this one on top of it?

AndrewJSchofield · 2023-11-23T09:04:00Z

@dajac I suggest you simply merge #14871 and let me deal with the fall-out. This one is not going to be far behind.

dajac · 2023-11-23T10:06:42Z

I just merged #14781. Thanks!

AndrewJSchofield · 2023-11-24T10:57:05Z

This PR now reflects the changes in KAFKA-14781 and also tests the new consumer.

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

dajac · 2023-11-24T16:54:08Z

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

    val numRecords = 10000
    val producer = createProducer()
    val startingTimestamp = System.currentTimeMillis()
    sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)

+    this.consumerConfig.setProperty(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)


Adding this change to all the tests is a bit annoying. Have you considered adding this to IntegrationTestHarness.doSetup or in createConsumer? We could infer it like we did with isNewGroupCoordinatorEnabled() in the same class.

I did consider that. I have already endured the annoyance. I'll take a look.

Can we remove this line now?

lucasbru

Scala 2.12 build needs to be fixed, and I left some minor comments. But generally this looks good to me

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

lucasbru · 2023-11-27T10:32:13Z

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

    val numRecords = 10000
    val producer = createProducer()
    val startingTimestamp = System.currentTimeMillis()
    sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)

+    this.consumerConfig.setProperty(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)


Can we remove this line now?

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

lucasbru · 2023-11-28T15:26:13Z

CI is blocked by a hanging test bug that is going to celebrate its fourth birthday soon: https://issues.apache.org/jira/browse/KAFKA-9470. I restarted CI and submitted #14855

lucasbru · 2023-11-28T20:25:40Z

I created 3 new flaky test tickets, but all failed tests were flaky before on master.

This commit parameterizes the consumer integration tests so they can be run against the existing "generic" group protocol and the new "consumer" group protocol introduced in KIP-848. The KIP-848 client code is under construction so some of the tests do not run on both variants to start with, but the idea is that the tests can be enabled as the gaps in functionality are closed. Reviewers: Lucas Brutschy <[email protected]>, Kirk True <[email protected]>

Merge pull request #2 from apache/trunk

a024310

Merge commits from trunk

AndrewJSchofield marked this pull request as draft November 14, 2023 17:02

kirktrue approved these changes Nov 14, 2023

View reviewed changes

AndrewJSchofield added 2 commits November 16, 2023 12:04

Merge branch 'apache:trunk' into trunk

88f2e32

Merge branch 'apache:trunk' into trunk

17eb4ef

AndrewJSchofield force-pushed the KAFKA-15544 branch from 2cec36d to 951e05d Compare November 20, 2023 16:44

AndrewJSchofield marked this pull request as ready for review November 20, 2023 16:58

AndrewJSchofield marked this pull request as draft November 21, 2023 09:05

AndrewJSchofield added 2 commits November 21, 2023 09:10

Merge branch 'apache:trunk' into trunk

b985806

Merge branch 'apache:trunk' into trunk

9ed0a44

AndrewJSchofield force-pushed the KAFKA-15544 branch from 951e05d to 54068f5 Compare November 21, 2023 13:52

AndrewJSchofield marked this pull request as ready for review November 21, 2023 13:52

cadonna added the ctr Consumer Threading Refactor (KIP-848) label Nov 21, 2023

cadonna reviewed Nov 22, 2023

View reviewed changes

AndrewJSchofield added 3 commits November 23, 2023 20:09

Merge branch 'apache:trunk' into trunk

d4196c3

KAFKA-15544: Enable integration tests new consumer

f8db168

Refactoring following KAFKA-14781 merge

4468494

AndrewJSchofield force-pushed the KAFKA-15544 branch from 54068f5 to 4468494 Compare November 24, 2023 10:55

AndrewJSchofield requested a review from cadonna November 24, 2023 10:56

dajac reviewed Nov 24, 2023

View reviewed changes

core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala Show resolved Hide resolved

core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala Outdated Show resolved Hide resolved

Test KIP-848 with generic group protocol

5fabfb3

dajac reviewed Nov 24, 2023

View reviewed changes

Set group protocol in test setup

ac05ca8

lucasbru reviewed Nov 27, 2023

View reviewed changes

AndrewJSchofield added 3 commits November 27, 2023 11:19

Fix Scala 2.12 method disambigution problem

1ef9c0d

Work around Scala version incompatibility

649b7a5

Review comments

d7064f8

lucasbru approved these changes Nov 28, 2023

View reviewed changes

lucasbru merged commit 161b94d into apache:trunk Nov 28, 2023
1 check failed

AndrewJSchofield deleted the KAFKA-15544 branch December 3, 2023 22:13

		@@ -66,6 +66,7 @@ abstract class AbstractConsumerTest extends BaseRequestTest {


		override protected def brokerPropertyOverrides(properties: Properties): Unit = {
		properties.setProperty(KafkaConfig.NewGroupCoordinatorEnableProp, "true") // enable KIP-848 group coordinator

		// No valid fetch position due to the absence of consumer.poll; and therefore no offset was committed to
		// tp. The committed offset should be null. This is intentional.

KAFKA-15544: Enable integration tests for new consumer #14758

KAFKA-15544: Enable integration tests for new consumer #14758

Conversation

AndrewJSchofield commented Nov 14, 2023

Committer Checklist (excluded from commit message)

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirktrue commented Nov 15, 2023

AndrewJSchofield commented Nov 21, 2023

kirktrue commented Nov 21, 2023

cadonna commented Nov 21, 2023

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dajac commented Nov 22, 2023

AndrewJSchofield commented Nov 23, 2023

dajac commented Nov 23, 2023

AndrewJSchofield commented Nov 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lucasbru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lucasbru commented Nov 28, 2023

lucasbru commented Nov 28, 2023