KAFKA-14499: [7/7] Add integration tests for OffsetCommit API and OffsetFetch API #14353

dajac · 2023-09-07T14:34:22Z

This patch adds integration tests for the OffsetCommit API and the OffsetFetch API. The tests runs against the old and the new group coordinator and with the new and the old consumer rebalance protocol.

Committer Checklist (excluded from commit message)

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

dajac · 2023-09-07T14:35:49Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

+            );
+        }
+
+        public Builder(String groupId,


Wire the member id/epoch in the builder.

dajac · 2023-09-07T14:36:00Z

clients/src/main/resources/common/message/OffsetFetchRequest.json

@@ -56,7 +56,7 @@
      "about": "Each group we would like to fetch offsets for", "fields": [
      { "name": "GroupId", "type": "string", "versions": "8+", "entityType": "groupId",
        "about": "The group ID."},
-      { "name": "MemberId", "type": "string", "versions": "9+", "nullableVersions": "9+", "default": null, "ignorable": true,
+      { "name": "MemberId", "type": "string", "versions": "9+", "nullableVersions": "9+", "default": "null", "ignorable": true,


Found this bug with the tests. Yeah!

dajac · 2023-09-07T14:36:12Z

core/src/main/scala/kafka/server/KafkaApis.scala

+        .setMemberId(offsetFetchRequest.memberId)
+        .setMemberEpoch(offsetFetchRequest.memberEpoch)


Also found this one. Yeah!

dajac · 2023-09-07T14:37:51Z

core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala

@@ -14,233 +14,532 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-
 package kafka.server


I rewrote all this test suite. The new one covers all the old cases as well, I think.

jolshan · 2023-09-07T17:45:33Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

                for (TopicPartition topicPartition : partitions) {
                    String topicName = topicPartition.topic();
-                    OffsetFetchRequestTopic topic = offsetFetchRequestTopicMap.getOrDefault(
-                        topicName, new OffsetFetchRequestTopic().setName(topicName));
+                    OffsetFetchRequestTopics topic = offsetFetchRequestTopicMap.getOrDefault(


If this object represents one topic, does it make sense for the name to be plural?

Yeah, this does not make sense but this is how the schema is defined. I can try to come up with a better name but I would do it separately.

jolshan · 2023-09-07T19:00:41Z

core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala

+import scala.jdk.CollectionConverters._
+import scala.reflect.ClassTag
+
+class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {


Is the goal here to create a new baseRequestTest class? I noticed we didn't extend the existing BaseRequestTest class

We can't reuse BaseRequestTest when using @ClusterTest. Here, I created GroupCoordinatorBaseRequestTest to group the common functionalities required to test Group Coordinator Apis.

I see. The naming reminds me of BaseRequest test and we have the same/similar connectAndReceive method. I think this is fine, but these test files do start to get confusing when we have so many similarly named but very different classes. Not something we need to address in this PR though.

Yeah, I agree. I was considering using a test context class instead of using a base class. I am not sure if it would really be better though... What do you think?

I think it is ok as it is. I wonder if naming helps? Either way we can do as a followup

jolshan · 2023-09-07T19:03:42Z

core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala

+  @ClusterTest(serverProperties = Array(
+    new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
+    new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+    new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),


Is this the current way we enable setting new consumer group protocol? Setting these two configs?

Yeah, at the moment, you have to use unstable.api.versions.enable and group.coordinator.new.enable. This will change when we release it though. It will be based on the metadata version.

Sorry I didn't ask this clearly. For testOffsetCommitWithNewConsumerGroupProtocolAndNewGroupCoordinator and testOffsetCommitWithOldConsumerGroupProtocolAndNewGroupCoordinator the only difference was the timeout configs:

new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000") new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000")

Is this how we enable the new group protocol? Both methods set unstable.api.versions.enable and group.coordinator.new.enable to be true. Should the second one have the api versions be false?

The two settings that you mentioned are regular configs for the new protocol. They don't enable the new protocol. As I said in my previous comment, the new protocol is automatically enabled when unstable.api.versions.enable and group.coordinator.new.enable are set at the moment. Later, the metadata version will be used.

I feel like maybe I'm missing something obvious but the test is called testOffsetCommitWithOldConsumerGroupProtocolAndNewGroupCoordinator() so it says we use the old protocol.

But

the new protocol is automatically enabled when unstable.api.versions.enable and group.coordinator.new.enable are set at the moment.

So should one of them be false so we have the old consumer group protocol?

def testOffsetCommitWithOldConsumerGroupProtocolAndNewGroupCoordinator(): Unit = { testOffsetCommit(false) }

Whether the new or the old protocol is used is defined by the parameter passed to testOffsetCommit. It is not used automatically because it is available. I need to set unstable.api.versions.enable here to test the unreleased versions of the offset commit/fetch apis. I know.. This is confusing...

Ok -- got it. Thanks for clarifying. That's what I was looking for.

jolshan · 2023-09-07T20:15:56Z

core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala

+    }
+
+    // Start from version 1 because version 0 goes to ZK.
+    for (version <- 1 to ApiKeys.OFFSET_COMMIT.latestVersion(isUnstableApiEnabled)) {


Should do we want to consider the empty group ID case here?

I can add it. It is similar for the unknown case.

Ah sorry -- maybe I was a bit unclear here too. Can we have the "null" group ID in this case? Or do we never have this case?

Got it. GroupId is not nullable so that would throw a NPE on the client side. This is the reason why I did not test it. This does not bring any value from an integration test point of view.

rreddy-22 · 2023-09-07T20:16:25Z

core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala

+      numPartitions = 3
+    )
+
+    // Create groups and commits offsets.


nit: 'commit' or 'creates'

jolshan · 2023-09-07T20:21:23Z

core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala

+        )
+      )
+
+      assertEquals(


nit: comment here this is the unknown group test case.

jolshan · 2023-09-07T20:23:00Z

core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala

+    testMultipleGroupsOffsetFetch(useNewProtocol = false, requireStable = true)
+  }
+
+  private def testSingleGroupOffsetFetch(useNewProtocol: Boolean, requireStable: Boolean): Unit = {


This method is a little large and contains quite a few smaller test cases within. Is there a way we can break it up a bit more?

Sure. I can separate the fetch offsets vs fetch all offsets cases.

jolshan · 2023-09-07T20:50:34Z

core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala

-    groups.foreach { groupId =>
-      consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId)
-      commitOffsets(partitionMap(groupId))
+  private def testMultipleGroupsOffsetFetch(useNewProtocol: Boolean, requireStable: Boolean): Unit = {


Another "my understanding" question. 😅 Do only admin clients request multiple group offsets at once? I probably should have realized this earlier, but I see we don't pass member id or epoch in this version of the request.

Yeah, that's correct.

dajac · 2023-09-08T08:56:09Z

@jolshan @rreddy-22 Thanks for your comments. I have addressed all of them if not mistaken. I have also fixed a few tests and extracted common code into the base class. Please have a second look a it.

jolshan

Thanks!

…setFetch API (apache#14353) This patch adds integration tests for the OffsetCommit API and the OffsetFetch API. The tests runs against the old and the new group coordinator and with the new and the old consumer rebalance protocol. Reviewers: Ritika Reddy <rreddy@confluent.io>, Justine Olshan <jolshan@confluent.io>

dajac added 3 commits September 6, 2023 18:24

Add integration tests for the OffsetCommit API

ae77d4c

Add integration tests for the OffsetFetch API

11ef4eb

cleanup

6d85c6e

dajac added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Sep 7, 2023

dajac commented Sep 7, 2023

View reviewed changes

fix

1b41b35

dajac commented Sep 7, 2023

View reviewed changes

fixup

8d7a599

jolshan reviewed Sep 7, 2023

View reviewed changes

rreddy-22 reviewed Sep 7, 2023

View reviewed changes

jolshan reviewed Sep 7, 2023

View reviewed changes

dajac added 4 commits September 8, 2023 10:11

fix tests & minor improvements

3c39b6f

address minor comments + add more test cases

00a459e

address minor comments

d6e8500

refactor

09038cd

refactor

604c9b3

dajac mentioned this pull request Sep 11, 2023

KAFKA-14503: Implement ListGroups #14271

Merged

3 tasks

jolshan approved these changes Sep 11, 2023

View reviewed changes

dajac merged commit 6123071 into apache:trunk Sep 11, 2023

dajac deleted the KAFKA-14499-7 branch September 11, 2023 17:48

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14499: [7/7] Add integration tests for OffsetCommit API and OffsetFetch API #14353

KAFKA-14499: [7/7] Add integration tests for OffsetCommit API and OffsetFetch API #14353

dajac commented Sep 7, 2023

dajac Sep 7, 2023

dajac Sep 7, 2023

dajac Sep 7, 2023

dajac Sep 7, 2023

jolshan Sep 7, 2023

dajac Sep 7, 2023

jolshan Sep 7, 2023

dajac Sep 8, 2023

jolshan Sep 8, 2023

dajac Sep 11, 2023

jolshan Sep 11, 2023

jolshan Sep 7, 2023

dajac Sep 8, 2023

jolshan Sep 8, 2023 •

edited

Loading

dajac Sep 11, 2023

jolshan Sep 11, 2023

dajac Sep 11, 2023

jolshan Sep 11, 2023

jolshan Sep 7, 2023

dajac Sep 8, 2023

jolshan Sep 8, 2023

dajac Sep 11, 2023 •

edited

Loading

rreddy-22 Sep 7, 2023

jolshan Sep 7, 2023

jolshan Sep 7, 2023

dajac Sep 8, 2023

jolshan Sep 7, 2023

dajac Sep 8, 2023

dajac commented Sep 8, 2023

jolshan left a comment

		.setMemberId(offsetFetchRequest.memberId)
		.setMemberEpoch(offsetFetchRequest.memberEpoch)

KAFKA-14499: [7/7] Add integration tests for OffsetCommit API and OffsetFetch API #14353

KAFKA-14499: [7/7] Add integration tests for OffsetCommit API and OffsetFetch API #14353

Conversation

dajac commented Sep 7, 2023

Committer Checklist (excluded from commit message)

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

jolshan Sep 8, 2023 • edited Loading

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 Sep 11, 2023 • edited Loading

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 Sep 8, 2023

jolshan left a comment

Choose a reason for hiding this comment

jolshan Sep 8, 2023 •

edited

Loading

dajac Sep 11, 2023 •

edited

Loading