showuon commented on a change in pull request #11019:
URL: https://github.com/apache/kafka/pull/11019#discussion_r669471550
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
##########
@@ -97,54 +108,79 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetDeleteResponse response = (OffsetDeleteResponse)
abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
+ final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
final Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleGroupError(groupId, error, failed, groupsToUnmap,
groupsToRetry);
} else {
- final Map<TopicPartition, Errors> partitions = new HashMap<>();
- response.data().topics().forEach(topic ->
+ final Map<TopicPartition, Errors> partitionResults = new
HashMap<>();
+ response.data().topics().forEach(topic ->
topic.partitions().forEach(partition -> {
Errors partitionError =
Errors.forCode(partition.errorCode());
- if (!handleError(groupId, partitionError, failed,
unmapped)) {
- partitions.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
- }
+
+ partitionResults.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
})
);
- if (!partitions.isEmpty())
- completed.put(groupId, partitions);
+
+ completed.put(groupId, partitionResults);
+ }
+
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ completed,
+ failed,
+ Collections.emptyList()
+ );
+ } else {
+ // retry the request, so don't send completed/failed results back
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
}
- return new ApiResult<>(completed, failed, unmapped);
}
- private boolean handleError(
+ private void handleGroupError(
CoordinatorKey groupId,
Errors error,
Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
case GROUP_ID_NOT_FOUND:
case INVALID_GROUP_ID:
- log.error("Received non retriable error for group {} in
`DeleteConsumerGroupOffsets` response", groupId,
- error.exception());
+ case NON_EMPTY_GROUP:
+ log.debug("`OffsetDelete` request for group id {} failed due
to error {}.", groupId, error);
Review comment:
Nice catch! I'll also update other PRs.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
