skaundinya15 commented on a change in pull request #10962:
URL: https://github.com/apache/kafka/pull/10962#discussion_r664738871
##########
File path:
clients/src/test/java/org/apache/kafka/common/requests/OffsetFetchRequestTest.java
##########
@@ -76,62 +73,169 @@ public void testConstructor() {
}
for (short version : ApiKeys.OFFSET_FETCH.allVersions()) {
- OffsetFetchRequest request = builder.build(version);
- assertFalse(request.isAllPartitions());
- assertEquals(groupId, request.groupId());
- assertEquals(partitions, request.partitions());
-
- OffsetFetchResponse response =
request.getErrorResponse(throttleTimeMs, Errors.NONE);
- assertEquals(Errors.NONE, response.error());
- assertFalse(response.hasError());
- assertEquals(Collections.singletonMap(Errors.NONE, version <=
(short) 1 ? 3 : 1), response.errorCounts(),
- "Incorrect error count for version " + version);
-
- if (version <= 1) {
- assertEquals(expectedData, response.responseData());
+ if (version < 8) {
+ builder = new OffsetFetchRequest.Builder(
+ group1,
+ false,
+ partitions,
+ false);
+ assertFalse(builder.isAllTopicPartitions());
+ OffsetFetchRequest request = builder.build(version);
+ assertFalse(request.isAllPartitions());
+ assertEquals(group1, request.groupId());
+ assertEquals(partitions, request.partitions());
+
+ OffsetFetchResponse response =
request.getErrorResponse(throttleTimeMs, Errors.NONE);
+ assertEquals(Errors.NONE, response.error());
+ assertFalse(response.hasError());
+ assertEquals(Collections.singletonMap(Errors.NONE, version <=
(short) 1 ? 3 : 1), response.errorCounts(),
+ "Incorrect error count for version " + version);
+
+ if (version <= 1) {
+ assertEquals(expectedData, response.responseDataV0ToV7());
+ }
+
+ if (version >= 3) {
+ assertEquals(throttleTimeMs, response.throttleTimeMs());
+ } else {
+ assertEquals(DEFAULT_THROTTLE_TIME,
response.throttleTimeMs());
+ }
+ } else {
+ builder = new Builder(Collections.singletonMap(group1,
partitions), false, false);
+ OffsetFetchRequest request = builder.build(version);
+ Map<String, List<TopicPartition>> groupToPartitionMap =
+ request.groupIdsToPartitions();
+ Map<String, List<OffsetFetchRequestTopics>> groupToTopicMap =
+ request.groupIdsToTopics();
+ assertFalse(request.isAllPartitionsForGroup(group1));
+ assertTrue(groupToPartitionMap.containsKey(group1) &&
groupToTopicMap.containsKey(
+ group1));
+ assertEquals(partitions, groupToPartitionMap.get(group1));
+ OffsetFetchResponse response =
request.getErrorResponse(throttleTimeMs, Errors.NONE);
+ assertEquals(Errors.NONE, response.groupLevelError(group1));
+ assertFalse(response.groupHasError(group1));
+ assertEquals(Collections.singletonMap(Errors.NONE, 1),
response.errorCounts(),
+ "Incorrect error count for version " + version);
+ assertEquals(throttleTimeMs, response.throttleTimeMs());
}
+ }
+ }
+
+ @Test
+ public void testConstructorWithMultipleGroups() {
+ List<TopicPartition> topic1Partitions = Arrays.asList(
+ new TopicPartition(topicOne, partitionOne),
+ new TopicPartition(topicOne, partitionTwo));
+ List<TopicPartition> topic2Partitions = Arrays.asList(
+ new TopicPartition(topicTwo, partitionOne),
+ new TopicPartition(topicTwo, partitionTwo));
+ List<TopicPartition> topic3Partitions = Arrays.asList(
+ new TopicPartition(topicThree, partitionOne),
+ new TopicPartition(topicThree, partitionTwo));
+ Map<String, List<TopicPartition>> groupToTp = new HashMap<>();
+ groupToTp.put(group1, topic1Partitions);
+ groupToTp.put(group2, topic2Partitions);
+ groupToTp.put(group3, topic3Partitions);
+ groupToTp.put(group4, null);
+ groupToTp.put(group5, null);
+ int throttleTimeMs = 10;
- if (version >= 3) {
+ for (short version : ApiKeys.OFFSET_FETCH.allVersions()) {
+ if (version >= 8) {
+ builder = new Builder(groupToTp, false, false);
+ OffsetFetchRequest request = builder.build(version);
+ Map<String, List<TopicPartition>> groupToPartitionMap =
+ request.groupIdsToPartitions();
+ Map<String, List<OffsetFetchRequestTopics>> groupToTopicMap =
+ request.groupIdsToTopics();
+ assertEquals(groupToTp.keySet(), groupToTopicMap.keySet());
+ assertEquals(groupToTp.keySet(), groupToPartitionMap.keySet());
+ assertFalse(request.isAllPartitionsForGroup(group1));
+ assertFalse(request.isAllPartitionsForGroup(group2));
+ assertFalse(request.isAllPartitionsForGroup(group3));
+ assertTrue(request.isAllPartitionsForGroup(group4));
+ assertTrue(request.isAllPartitionsForGroup(group5));
+ OffsetFetchResponse response =
request.getErrorResponse(throttleTimeMs, Errors.NONE);
+ assertEquals(Errors.NONE, response.groupLevelError(group1));
+ assertEquals(Errors.NONE, response.groupLevelError(group2));
+ assertEquals(Errors.NONE, response.groupLevelError(group3));
+ assertEquals(Errors.NONE, response.groupLevelError(group4));
+ assertEquals(Errors.NONE, response.groupLevelError(group5));
+ assertFalse(response.groupHasError(group1));
+ assertFalse(response.groupHasError(group2));
+ assertFalse(response.groupHasError(group3));
+ assertFalse(response.groupHasError(group4));
+ assertFalse(response.groupHasError(group5));
+ assertEquals(Collections.singletonMap(Errors.NONE, 5),
response.errorCounts(),
+ "Incorrect error count for version " + version);
assertEquals(throttleTimeMs, response.throttleTimeMs());
- } else {
- assertEquals(DEFAULT_THROTTLE_TIME, response.throttleTimeMs());
}
}
}
@Test
- public void testConstructorFailForUnsupportedRequireStable() {
+ public void testBuildThrowForUnsupportedBatchRequest() {
for (short version : ApiKeys.OFFSET_FETCH.allVersions()) {
- // The builder needs to be initialized every cycle as the internal
data `requireStable` flag is flipped.
- builder = new OffsetFetchRequest.Builder(groupId, true, null,
false);
- final short finalVersion = version;
- if (version < 2) {
- assertThrows(UnsupportedVersionException.class, () ->
builder.build(finalVersion));
- } else {
- OffsetFetchRequest request = builder.build(finalVersion);
- assertEquals(groupId, request.groupId());
- assertNull(request.partitions());
- assertTrue(request.isAllPartitions());
- if (version < 7) {
- assertFalse(request.requireStable());
- } else {
- assertTrue(request.requireStable());
- }
+ if (version < 8) {
+ Map<String, List<TopicPartition>> groupPartitionMap = new
HashMap<>();
+ groupPartitionMap.put(group1, null);
+ groupPartitionMap.put(group2, null);
+ builder = new Builder(groupPartitionMap, true, false);
+ final short finalVersion = version;
+ assertThrows(NoBatchedOffsetFetchRequestException.class, () ->
builder.build(finalVersion));
Review comment:
This specific test checks if the correct exception is thrown if we try
to create a batched request for a version less than 8. The `testConstructor`
method tests the single group request with version 8 and above. Is there any
other case you think we should add for `version >=8`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
