showuon commented on a change in pull request #11022:
URL: https://github.com/apache/kafka/pull/11022#discussion_r670073886
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java
##########
@@ -151,38 +154,45 @@ public String apiName() {
completed.put(groupIdKey, consumerGroupDescription);
} else {
failed.put(groupIdKey, new IllegalArgumentException(
- String.format("GroupId %s is not a consumer group
(%s).",
- groupIdKey.idValue, protocolType)));
+ String.format("GroupId %s is not a consumer group (%s).",
+ groupIdKey.idValue, protocolType)));
}
}
- return new ApiResult<>(completed, failed, unmapped);
+
+ return new ApiResult<>(completed, failed, new
ArrayList<>(groupsToUnmap));
}
private void handleError(
CoordinatorKey groupId,
Errors error,
Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in
`DescribeGroups` response", groupId,
- error.exception());
+ log.debug("`DescribeGroups` request for group id {} failed due
to error {}", groupId.idValue, error);
failed.put(groupId, error.exception());
break;
case COORDINATOR_LOAD_IN_PROGRESS:
- case COORDINATOR_NOT_AVAILABLE:
+ // 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);
+ groupsToRetry.add(groupId);
break;
+ case COORDINATOR_NOT_AVAILABLE:
case NOT_COORDINATOR:
- log.debug("DescribeGroups request for group {} returned error
{}. Will retry",
- groupId, error);
- unmapped.add(groupId);
+ // 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);
+ groupsToUnmap.add(groupId);
break;
default:
- log.error("Received unexpected error for group {} in
`DescribeGroups` response",
- groupId, error.exception());
- failed.put(groupId, error.exception(
- "Received unexpected error for group " + groupId + "
in `DescribeGroups` response"));
+ final String unexpectedErrorMsg =
+ String.format("`DescribeGroups` request for group id %s
failed due to error %s", groupId.idValue, error);
+ log.error(unexpectedErrorMsg);
+ failed.put(groupId, error.exception(unexpectedErrorMsg));
Review comment:
Agree. I just follow the previous behavior. Remove the error message.
Thanks.
--
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
