lucasbru commented on code in PR #19116:
URL: https://github.com/apache/kafka/pull/19116#discussion_r1984784869
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -5762,6 +5764,347 @@ public void
testDeleteStreamsGroupOffsetsFindCoordinatorNonRetriableErrors() thr
}
}
+ @Test
+ public void testDescribeStreamsGroups() throws Exception {
+ try (AdminClientUnitTestEnv env = new
AdminClientUnitTestEnv(mockCluster(1, 0))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+
+ // Retriable FindCoordinatorResponse errors should be retried
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE,
Node.noNode()));
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.COORDINATOR_LOAD_IN_PROGRESS,
Node.noNode()));
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE,
env.cluster().controller()));
+
+ StreamsGroupDescribeResponseData data = new
StreamsGroupDescribeResponseData();
+
+ // Retriable errors should be retried
+ data.groups().add(new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(GROUP_ID)
+ .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code()));
+ env.kafkaClient().prepareResponse(new
StreamsGroupDescribeResponse(data));
+
+ /*
Review Comment:
Done
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -5762,6 +5764,347 @@ public void
testDeleteStreamsGroupOffsetsFindCoordinatorNonRetriableErrors() thr
}
}
+ @Test
+ public void testDescribeStreamsGroups() throws Exception {
+ try (AdminClientUnitTestEnv env = new
AdminClientUnitTestEnv(mockCluster(1, 0))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+
+ // Retriable FindCoordinatorResponse errors should be retried
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE,
Node.noNode()));
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.COORDINATOR_LOAD_IN_PROGRESS,
Node.noNode()));
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE,
env.cluster().controller()));
+
+ StreamsGroupDescribeResponseData data = new
StreamsGroupDescribeResponseData();
+
+ // Retriable errors should be retried
+ data.groups().add(new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(GROUP_ID)
+ .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code()));
+ env.kafkaClient().prepareResponse(new
StreamsGroupDescribeResponse(data));
+
+ /*
+ * We need to return two responses here, one with NOT_COORDINATOR
error when calling describe streams group
+ * api using coordinator that has moved. This will retry whole
operation. So we need to again respond with a
+ * FindCoordinatorResponse.
+ *
+ * And the same reason for COORDINATOR_NOT_AVAILABLE error response
+ */
+ data = new StreamsGroupDescribeResponseData();
+ data.groups().add(new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(GROUP_ID)
+ .setErrorCode(Errors.NOT_COORDINATOR.code()));
+ env.kafkaClient().prepareResponse(new
StreamsGroupDescribeResponse(data));
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE,
env.cluster().controller()));
+
+ data = new StreamsGroupDescribeResponseData();
+ data.groups().add(new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(GROUP_ID)
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()));
+ env.kafkaClient().prepareResponse(new
StreamsGroupDescribeResponse(data));
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE,
env.cluster().controller()));
+
+ data = makeFullStreamsGroupDescribeResponse();
+
+ env.kafkaClient().prepareResponse(new
StreamsGroupDescribeResponse(data));
+
+ final DescribeStreamsGroupsResult result =
env.adminClient().describeStreamsGroups(singletonList(GROUP_ID));
+ final StreamsGroupDescription groupDescription =
result.describedGroups().get(GROUP_ID).get();
+
+ final String subtopologyId = "my_subtopology";
+ StreamsGroupMemberAssignment.TaskIds expectedActiveTasks1 =
+ new StreamsGroupMemberAssignment.TaskIds(subtopologyId,
asList(0, 1, 2));
+ StreamsGroupMemberAssignment.TaskIds expectedStandbyTasks1 =
+ new StreamsGroupMemberAssignment.TaskIds(subtopologyId,
asList(3, 4, 5));
+ StreamsGroupMemberAssignment.TaskIds expectedWarmupTasks1 =
+ new StreamsGroupMemberAssignment.TaskIds(subtopologyId,
asList(6, 7, 8));
+ StreamsGroupMemberAssignment.TaskIds expectedActiveTasks2 =
+ new StreamsGroupMemberAssignment.TaskIds(subtopologyId,
asList(3, 4, 5));
+ StreamsGroupMemberAssignment.TaskIds expectedStandbyTasks2 =
+ new StreamsGroupMemberAssignment.TaskIds(subtopologyId,
asList(6, 7, 8));
+ StreamsGroupMemberAssignment.TaskIds expectedWarmupTasks2 =
+ new StreamsGroupMemberAssignment.TaskIds(subtopologyId,
asList(0, 1, 2));
+ StreamsGroupMemberAssignment expectedMemberAssignment = new
StreamsGroupMemberAssignment(
+ singletonList(expectedActiveTasks1),
+ singletonList(expectedStandbyTasks1),
+ singletonList(expectedWarmupTasks1)
+ );
+ StreamsGroupMemberAssignment expectedTargetAssignment = new
StreamsGroupMemberAssignment(
+ singletonList(expectedActiveTasks2),
+ singletonList(expectedStandbyTasks2),
+ singletonList(expectedWarmupTasks2)
+ );
+ final String instanceId = "instance-id";
+ final String rackId = "rack-id";
+ StreamsGroupMemberDescription expectedMemberOne = new
StreamsGroupMemberDescription(
+ "0",
+ 1,
+ Optional.of(instanceId),
+ Optional.of(rackId),
+ "clientId0",
+ "clientHost",
+ 0,
+ "processId",
+ Optional.of(new
StreamsGroupMemberDescription.Endpoint("localhost", 8080)),
+ Collections.singletonMap("key", "value"),
+ Collections.singletonList(new
StreamsGroupMemberDescription.TaskOffset(subtopologyId, 0, 0)),
+ Collections.singletonList(new
StreamsGroupMemberDescription.TaskOffset(subtopologyId, 0, 1)),
+ expectedMemberAssignment,
+ expectedTargetAssignment,
+ true
+ );
+
+ StreamsGroupMemberDescription expectedMemberTwo = new
StreamsGroupMemberDescription(
+ "1",
+ 2,
+ Optional.empty(),
+ Optional.empty(),
+ "clientId1",
+ "clientHost",
+ 1,
+ "processId2",
+ Optional.empty(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new StreamsGroupMemberAssignment(Collections.emptyList(),
Collections.emptyList(), Collections.emptyList()),
+ new StreamsGroupMemberAssignment(Collections.emptyList(),
Collections.emptyList(), Collections.emptyList()),
+ false
+ );
+
+ StreamsGroupSubtopologyDescription expectedSubtopologyDescription
= new StreamsGroupSubtopologyDescription(
+ subtopologyId,
+ Collections.singletonList("my_source_topic"),
+ Collections.singletonList("my_repartition_sink_topic"),
+ Collections.singletonMap(
+ "my_changelog_topic",
+ new StreamsGroupSubtopologyDescription.TopicInfo(
+ 0,
+ (short) 3,
+ Collections.singletonMap("key1", "value1")
+ )
+ ),
+ Collections.singletonMap(
+ "my_repartition_topic",
+ new StreamsGroupSubtopologyDescription.TopicInfo(
+ 99,
+ (short) 0,
+ Collections.emptyMap()
+ )
+ )
+ );
+
+ assertEquals(1, result.describedGroups().size());
+ assertEquals(GROUP_ID, groupDescription.groupId());
+ assertEquals(2, groupDescription.members().size());
+ Iterator<StreamsGroupMemberDescription> members =
groupDescription.members().iterator();
+ assertEquals(expectedMemberOne, members.next());
+ assertEquals(expectedMemberTwo, members.next());
+ assertEquals(1, groupDescription.subtopologies().size());
+ assertEquals(expectedSubtopologyDescription,
groupDescription.subtopologies().iterator().next());
+ assertEquals(2, groupDescription.groupEpoch());
+ assertEquals(1, groupDescription.targetAssignmentEpoch());
+
+ }
+ }
+
+ private static StreamsGroupDescribeResponseData
makeFullStreamsGroupDescribeResponse() {
Review Comment:
Done
--
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
