KAFKA-19440: Handle top-level errors in AlterShareGroupOffsets RPC #20049
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 |
|---|---|---|
|
|
@@ -21,8 +21,8 @@ | |
| import org.apache.kafka.clients.admin.KafkaAdminClient; | ||
| import org.apache.kafka.common.Node; | ||
| import org.apache.kafka.common.TopicPartition; | ||
| import org.apache.kafka.common.errors.ApiException; | ||
| import org.apache.kafka.common.message.AlterShareGroupOffsetsRequestData; | ||
| import org.apache.kafka.common.protocol.Errors; | ||
| import org.apache.kafka.common.requests.AbstractResponse; | ||
| import org.apache.kafka.common.requests.AlterShareGroupOffsetsRequest; | ||
|
|
@@ -33,7 +33,6 @@ | |
| import org.slf4j.Logger; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.HashSet; | ||
| import java.util.Map; | ||
|
|
@@ -42,7 +41,7 @@ | |
| /** | ||
| * This class is the handler for {@link KafkaAdminClient#alterShareGroupOffsets(String, Map, AlterShareGroupOffsetsOptions)} call | ||
| */ | ||
| public class AlterShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, ApiException>> { | ||
|
|
||
| private final CoordinatorKey groupId; | ||
|
|
||
|
|
@@ -52,16 +51,22 @@ public class AlterShareGroupOffsetsHandler extends AdminApiHandler.Batched<Coord | |
|
|
||
| private final CoordinatorStrategy lookupStrategy; | ||
|
|
||
| public AlterShareGroupOffsetsHandler(String groupId, Map<TopicPartition, Long> offsets, LogContext logContext) { | ||
| this.groupId = CoordinatorKey.byGroupId(groupId); | ||
| this.offsets = offsets; | ||
| this.log = logContext.logger(AlterShareGroupOffsetsHandler.class); | ||
| this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext); | ||
| } | ||
|
|
||
| public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, ApiException>> newFuture(String groupId) { | ||
| return AdminApiFuture.forKeys(Set.of(CoordinatorKey.byGroupId(groupId))); | ||
| } | ||
|
|
||
| private void validateKeys(Set<CoordinatorKey> groupIds) { | ||
| if (!groupIds.equals(Set.of(groupId))) { | ||
| throw new IllegalArgumentException("Received unexpected group ids " + groupIds + | ||
| " (expected only " + Set.of(groupId) + ")"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -87,30 +92,38 @@ public String apiName() { | |
| } | ||
|
|
||
| @Override | ||
| public ApiResult<CoordinatorKey, Map<TopicPartition, ApiException>> handleResponse(Node broker, Set<CoordinatorKey> keys, AbstractResponse abstractResponse) { | ||
| validateKeys(keys); | ||
|
|
||
| AlterShareGroupOffsetsResponse response = (AlterShareGroupOffsetsResponse) abstractResponse; | ||
| final Set<CoordinatorKey> groupsToUnmap = new HashSet<>(); | ||
| final Set<CoordinatorKey> groupsToRetry = new HashSet<>(); | ||
| final Map<TopicPartition, ApiException> partitionResults = new HashMap<>(); | ||
|
|
||
| if (response.data().errorCode() != Errors.NONE.code()) { | ||
| final Errors topLevelError = Errors.forCode(response.data().errorCode()); | ||
| final String topLevelErrorMessage = response.data().errorMessage(); | ||
|
|
||
| offsets.forEach((topicPartition, offset) -> | ||
| handleError( | ||
| groupId, | ||
| topicPartition, | ||
| topLevelError, | ||
| topLevelErrorMessage, | ||
| partitionResults, | ||
| groupsToUnmap, | ||
| groupsToRetry | ||
| )); | ||
| } else { | ||
| response.data().responses().forEach(topic -> topic.partitions().forEach(partition -> { | ||
| if (partition.errorCode() != Errors.NONE.code()) { | ||
| final Errors partitionError = Errors.forCode(partition.errorCode()); | ||
|
Member
There was a problem hiding this comment. nit: the debug message should use placeholder for
Member
There was a problem hiding this comment. for example: var partitionError = Errors.forCode(partition.errorCode());
var partitionErrorMessage = partition.errorMessage();
if (partitionError != Errors.NONE) {
log.debug("AlterShareGroupOffsets request for group id {} and topic-partition {}-{} failed and returned error {}. {}",
groupId.idValue, topic.topicName(), partition.partitionIndex(), partitionError, partitionErrorMessage);
}
partitionResults.put(new TopicPartition(topic.topicName(), partition.partitionIndex()), partitionError.exception(partitionErrorMessage));
Member
Author
There was a problem hiding this comment. Thanks @chia7712 . I'll put in a minor PR today. |
||
| final String partitionErrorMessage = partition.errorMessage(); | ||
| log.debug("AlterShareGroupOffsets request for group id {} and topic-partition {}-{} failed and returned error {}." + partitionErrorMessage, | ||
| groupId.idValue, topic.topicName(), partition.partitionIndex(), partitionError); | ||
| } | ||
| partitionResults.put(new TopicPartition(topic.topicName(), partition.partitionIndex()), Errors.forCode(partition.errorCode()).exception(partition.errorMessage())); | ||
| })); | ||
| } | ||
|
|
||
| if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) { | ||
|
|
@@ -121,23 +134,23 @@ public ApiResult<CoordinatorKey, Map<TopicPartition, Errors>> handleResponse(Nod | |
| } | ||
|
|
||
| private void handleError( | ||
| CoordinatorKey groupId, | ||
| TopicPartition topicPartition, | ||
| Errors error, | ||
| String errorMessage, | ||
| Map<TopicPartition, ApiException> partitionResults, | ||
| Set<CoordinatorKey> groupsToUnmap, | ||
| Set<CoordinatorKey> groupsToRetry | ||
| ) { | ||
| switch (error) { | ||
| case COORDINATOR_LOAD_IN_PROGRESS: | ||
| case REBALANCE_IN_PROGRESS: | ||
| log.debug("AlterShareGroupOffsets request for group id {} returned error {}. Will retry." + errorMessage, groupId.idValue, error); | ||
| groupsToRetry.add(groupId); | ||
| break; | ||
| case COORDINATOR_NOT_AVAILABLE: | ||
| case NOT_COORDINATOR: | ||
| log.debug("AlterShareGroupOffsets request for group id {} returned error {}. Will rediscover the coordinator and retry." + errorMessage, | ||
| groupId.idValue, error); | ||
| groupsToUnmap.add(groupId); | ||
| break; | ||
|
|
@@ -147,14 +160,12 @@ private void handleError( | |
| case UNKNOWN_SERVER_ERROR: | ||
| case KAFKA_STORAGE_ERROR: | ||
| case GROUP_AUTHORIZATION_FAILED: | ||
| log.debug("AlterShareGroupOffsets request for group id {} failed due to error {}." + errorMessage, groupId.idValue, error); | ||
| partitionResults.put(topicPartition, error.exception(errorMessage)); | ||
| break; | ||
| default: | ||
| log.error("AlterShareGroupOffsets request for group id {} failed due to unexpected error {}." + errorMessage, groupId.idValue, error); | ||
| partitionResults.put(topicPartition, error.exception(errorMessage)); | ||
| } | ||
| } | ||
|
|
||
|
|
||
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.
Thanks for the patch, just out of curiosity, would the AlterConsumerGroupOffsets RPC have the same issue?
kafka/clients/src/main/java/org/apache/kafka/clients/admin/AlterConsumerGroupOffsetsResult.java
Lines 68 to 79 in 05b6e81
Uh oh!
There was an error while loading. Please reload this page.
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.
I'll check it out before I merge, but the important difference here is in
KafkaApis.scala. For DeleteSGO, it already handled a non-zero error code. For AlterSGO, that code was missing.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.
Ah, sorry. I didn't read your comment accurately. The difference is that AlterShareGroupOffsets can successfully pass back an error code, which is why this is in terms of
ApiExceptionrather thanErrors. ForAlterConsumerGroupOffsets, the RPC is actuallyOffsetCommitand this does not have anErrorMessagefield at all. So, it cannot be fixed for consumer groups until we have a version bump on theOffsetCommitRPC.