AndrewJSchofield commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1473234152
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##########
@@ -158,36 +295,71 @@ public ApiResult<CoordinatorKey,
ConsumerGroupDescription> handleResponse(
return new ApiResult<>(completed, failed, new
ArrayList<>(groupsToUnmap));
}
+ private Set<TopicPartition>
convertAssignment(ConsumerGroupDescribeResponseData.Assignment assignment) {
+ return assignment.topicPartitions().stream().flatMap(topic ->
+ topic.partitions().stream().map(partition ->
+ new TopicPartition(topic.topicName(), partition)
+ )
+ ).collect(Collectors.toSet());
+ }
+
private void handleError(
CoordinatorKey groupId,
Errors error,
+ String errorMsg,
Map<CoordinatorKey, Throwable> failed,
- Set<CoordinatorKey> groupsToUnmap
+ Set<CoordinatorKey> groupsToUnmap,
+ boolean isConsumerGroupResponse
) {
+ String apiName = isConsumerGroupResponse ? "ConsumerGroupDescribe" :
"DescribeGroups";
+
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
- log.debug("`DescribeGroups` request for group id {} failed due
to error {}", groupId.idValue, error);
- failed.put(groupId, error.exception());
+ log.debug("`{}` request for group id {} failed due to error
{}.", apiName, groupId.idValue, error);
+ failed.put(groupId, error.exception(errorMsg));
break;
case COORDINATOR_LOAD_IN_PROGRESS:
// If the coordinator is in the middle of loading, then we
just need to retry
- log.debug("`DescribeGroups` request for group id {} failed
because the coordinator " +
- "is still in the process of loading state. Will retry",
groupId.idValue);
+ log.debug("`{}` request for group id {} failed because the
coordinator " +
+ "is still in the process of loading state. Will retry.",
apiName, groupId.idValue);
break;
case COORDINATOR_NOT_AVAILABLE:
case NOT_COORDINATOR:
// If the coordinator is unavailable or there was a
coordinator change, then we unmap
// the key so that we retry the `FindCoordinator` request
- log.debug("`DescribeGroups` request for group id {} returned
error {}. " +
- "Will attempt to find the coordinator again and retry",
groupId.idValue, error);
+ log.debug("`{}` request for group id {} returned error {}. " +
+ "Will attempt to find the coordinator again and retry.",
apiName, groupId.idValue, error);
groupsToUnmap.add(groupId);
break;
+ case UNSUPPORTED_VERSION:
+ if (isConsumerGroupResponse) {
+ log.debug("`{}` request for group id {} failed because the
API is not " +
+ "supported. Will retry with `DescribeGroups` API.",
apiName, groupId.idValue);
+ useClassicGroupApi.add(groupId.idValue);
+ } else {
+ log.error("`{}` request for group id {} because the
`ConsumerGroupDescribe` API is not supported.",
+ apiName, groupId.idValue);
+ failed.put(groupId, error.exception(errorMsg));
+ }
+ break;
+
+ case GROUP_ID_NOT_FOUND:
+ if (isConsumerGroupResponse) {
+ log.debug("`{}` request for group id {} failed because the
group is not " +
+ "a new consumer group. Will retry with
`DescribeGroups` API.", apiName, groupId.idValue);
+ useClassicGroupApi.add(groupId.idValue);
+ } else {
+ log.error("`{}` request for group id {} because the group
does not exist.", apiName, groupId.idValue);
Review Comment:
This doesn't seem grammatical. It would say "DescribeGroups request for
group id {} because the group does not exist". I suggest "failed" or something
like that is missing.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##########
@@ -158,36 +295,71 @@ public ApiResult<CoordinatorKey,
ConsumerGroupDescription> handleResponse(
return new ApiResult<>(completed, failed, new
ArrayList<>(groupsToUnmap));
}
+ private Set<TopicPartition>
convertAssignment(ConsumerGroupDescribeResponseData.Assignment assignment) {
+ return assignment.topicPartitions().stream().flatMap(topic ->
+ topic.partitions().stream().map(partition ->
+ new TopicPartition(topic.topicName(), partition)
+ )
+ ).collect(Collectors.toSet());
+ }
+
private void handleError(
CoordinatorKey groupId,
Errors error,
+ String errorMsg,
Map<CoordinatorKey, Throwable> failed,
- Set<CoordinatorKey> groupsToUnmap
+ Set<CoordinatorKey> groupsToUnmap,
+ boolean isConsumerGroupResponse
) {
+ String apiName = isConsumerGroupResponse ? "ConsumerGroupDescribe" :
"DescribeGroups";
+
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
- log.debug("`DescribeGroups` request for group id {} failed due
to error {}", groupId.idValue, error);
- failed.put(groupId, error.exception());
+ log.debug("`{}` request for group id {} failed due to error
{}.", apiName, groupId.idValue, error);
+ failed.put(groupId, error.exception(errorMsg));
break;
case COORDINATOR_LOAD_IN_PROGRESS:
// If the coordinator is in the middle of loading, then we
just need to retry
- log.debug("`DescribeGroups` request for group id {} failed
because the coordinator " +
- "is still in the process of loading state. Will retry",
groupId.idValue);
+ log.debug("`{}` request for group id {} failed because the
coordinator " +
+ "is still in the process of loading state. Will retry.",
apiName, groupId.idValue);
break;
case COORDINATOR_NOT_AVAILABLE:
case NOT_COORDINATOR:
// If the coordinator is unavailable or there was a
coordinator change, then we unmap
// the key so that we retry the `FindCoordinator` request
- log.debug("`DescribeGroups` request for group id {} returned
error {}. " +
- "Will attempt to find the coordinator again and retry",
groupId.idValue, error);
+ log.debug("`{}` request for group id {} returned error {}. " +
+ "Will attempt to find the coordinator again and retry.",
apiName, groupId.idValue, error);
groupsToUnmap.add(groupId);
break;
+ case UNSUPPORTED_VERSION:
+ if (isConsumerGroupResponse) {
+ log.debug("`{}` request for group id {} failed because the
API is not " +
+ "supported. Will retry with `DescribeGroups` API.",
apiName, groupId.idValue);
+ useClassicGroupApi.add(groupId.idValue);
+ } else {
+ log.error("`{}` request for group id {} because the
`ConsumerGroupDescribe` API is not supported.",
Review Comment:
I think this is missing at least one word in the log message.
--
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
