KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly. #11016
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,10 +16,12 @@ | |
| */ | ||
| package org.apache.kafka.clients.admin.internals; | ||
|
|
||
| import static java.util.Collections.singleton; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.HashSet; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
|
|
||
|
|
@@ -73,29 +75,38 @@ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicParti | |
| return AdminApiFuture.forKeys(Collections.singleton(CoordinatorKey.byGroupId(groupId))); | ||
| } | ||
|
|
||
| private void validateKeys(Set<CoordinatorKey> groupIds) { | ||
| if (!groupIds.equals(singleton(groupId))) { | ||
| throw new IllegalArgumentException("Received unexpected group ids " + groupIds + | ||
| " (expected only " + singleton(groupId) + ")"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public OffsetCommitRequest.Builder buildRequest( | ||
| int coordinatorId, | ||
| Set<CoordinatorKey> groupIds | ||
| ) { | ||
| validateKeys(groupIds); | ||
|
|
||
| Map<String, OffsetCommitRequestTopic> offsetData = new HashMap<>(); | ||
| offsets.forEach((topicPartition, offsetAndMetadata) -> { | ||
| OffsetCommitRequestTopic topic = offsetData.computeIfAbsent( | ||
| topicPartition.topic(), | ||
| key -> new OffsetCommitRequestTopic().setName(topicPartition.topic()) | ||
| ); | ||
|
|
||
| topic.partitions().add(new OffsetCommitRequestPartition() | ||
| .setCommittedOffset(offsetAndMetadata.offset()) | ||
| .setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch().orElse(-1)) | ||
| .setCommittedMetadata(offsetAndMetadata.metadata()) | ||
| .setPartitionIndex(topicPartition.partition())); | ||
| }); | ||
|
|
||
| OffsetCommitRequestData data = new OffsetCommitRequestData() | ||
| .setGroupId(groupId.idValue) | ||
| .setTopics(new ArrayList<>(offsetData.values())); | ||
|
|
||
| return new OffsetCommitRequest.Builder(data); | ||
| } | ||
|
|
||
|
|
@@ -105,53 +116,88 @@ public ApiResult<CoordinatorKey, Map<TopicPartition, Errors>> handleResponse( | |
| Set<CoordinatorKey> groupIds, | ||
| AbstractResponse abstractResponse | ||
| ) { | ||
| validateKeys(groupIds); | ||
|
|
||
| final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse; | ||
| final Set<CoordinatorKey> groupsToUnmap = new HashSet<>(); | ||
| final Set<CoordinatorKey> groupsToRetry = new HashSet<>(); | ||
| final Map<TopicPartition, Errors> partitionResults = new HashMap<>(); | ||
|
|
||
| for (OffsetCommitResponseTopic topic : response.data().topics()) { | ||
| for (OffsetCommitResponsePartition partition : topic.partitions()) { | ||
| TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex()); | ||
| Errors error = Errors.forCode(partition.errorCode()); | ||
|
|
||
| if (error != Errors.NONE) { | ||
| handleError( | ||
| groupId, | ||
| topicPartition, | ||
| error, | ||
| partitionResults, | ||
| groupsToUnmap, | ||
| groupsToRetry | ||
| ); | ||
| } else { | ||
| partitionResults.put(topicPartition, error); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) { | ||
| return ApiResult.completed(groupId, partitionResults); | ||
| } else { | ||
| return ApiResult.unmapped(new ArrayList<>(groupsToUnmap)); | ||
| } | ||
| } | ||
|
|
||
| private void handleError( | ||
| CoordinatorKey groupId, | ||
| TopicPartition topicPartition, | ||
| Errors error, | ||
| Map<TopicPartition, Errors> partitionResults, | ||
| Set<CoordinatorKey> groupsToUnmap, | ||
| Set<CoordinatorKey> groupsToRetry | ||
| ) { | ||
| switch (error) { | ||
| // If the coordinator is in the middle of loading, then we just need to retry. | ||
| case COORDINATOR_LOAD_IN_PROGRESS: | ||
| log.debug("OffsetCommit request for group id {} failed because the coordinator" + | ||
| " is still in the process of loading state. Will retry.", groupId.idValue); | ||
| groupsToRetry.add(groupId); | ||
| break; | ||
|
|
||
| // If the coordinator is not available, then we unmap and retry. | ||
| case COORDINATOR_NOT_AVAILABLE: | ||
| case NOT_COORDINATOR: | ||
| log.debug("OffsetCommit request for group id {} returned error {}. Will retry.", | ||
| groupId.idValue, error); | ||
| groupsToUnmap.add(groupId); | ||
| break; | ||
|
|
||
| // Group level errors. | ||
| case INVALID_GROUP_ID: | ||
| case REBALANCE_IN_PROGRESS: | ||
| case INVALID_COMMIT_OFFSET_SIZE: | ||
|
Contributor
There was a problem hiding this comment. Is this a group-level error?
Member
Author
There was a problem hiding this comment. I think it is. It basically indicate that we could write the group metadata to the log so it concerns the group. https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L448 |
||
| case GROUP_AUTHORIZATION_FAILED: | ||
| log.debug("OffsetCommit request for group id {} failed due to error {}.", | ||
| groupId.idValue, error); | ||
| partitionResults.put(topicPartition, error); | ||
|
Contributor
There was a problem hiding this comment. Should REBALANCE_IN_PROGRESS and GROUP_AUTHORIZATION_FAILED be added to
Member
Author
There was a problem hiding this comment. Yeah, that's a good question. Prior to KIP-599, we considered them as non retryable errors so I sticked to this here. I think that it might be a good idea to consider them as retryable errors but we should do it consistently for all the group handlers. How about filing a Jira for this and tackling it separately?
Contributor
There was a problem hiding this comment. Yes, we can open a JIRA to do it later.
Member
Author
There was a problem hiding this comment. |
||
| break; | ||
|
|
||
| // TopicPartition level errors. | ||
| case UNKNOWN_TOPIC_OR_PARTITION: | ||
| case OFFSET_METADATA_TOO_LARGE: | ||
| case TOPIC_AUTHORIZATION_FAILED: | ||
| log.debug("OffsetCommit request for group id {} and partition {} failed due" + | ||
| " to error {}.", groupId.idValue, topicPartition, error); | ||
| partitionResults.put(topicPartition, error); | ||
| break; | ||
|
|
||
| // Unexpected errors. | ||
| default: | ||
| log.error("OffsetCommit request for group id {} and partition {} failed due" + | ||
| " to unexpected error {}.", groupId.idValue, topicPartition, error); | ||
| partitionResults.put(topicPartition, error); | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is just a change to lambda expression, no content change, right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Right.