You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/07/10 16:47:33 UTC
[GitHub] [kafka] dajac opened a new pull request #11016: [WIP] Update `AlterConsumerGroupOffsetsHandler` to handle errors correctly.
dajac opened a new pull request #11016:
URL: https://github.com/apache/kafka/pull/11016
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rajinisivaram commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
rajinisivaram commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672176498
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -73,29 +75,40 @@ public String apiName() {
return AdminApiFuture.forKeys(Collections.singleton(CoordinatorKey.byGroupId(groupId)));
}
- @Override
- public OffsetCommitRequest.Builder buildRequest(int coordinatorId, Set<CoordinatorKey> keys) {
- List<OffsetCommitRequestTopic> topics = new ArrayList<>();
- Map<String, List<OffsetCommitRequestPartition>> offsetData = new HashMap<>();
- for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
- String topic = entry.getKey().topic();
- OffsetAndMetadata oam = entry.getValue();
- OffsetCommitRequestPartition partition = new OffsetCommitRequestPartition()
- .setCommittedOffset(oam.offset())
- .setCommittedLeaderEpoch(oam.leaderEpoch().orElse(-1))
- .setCommittedMetadata(oam.metadata())
- .setPartitionIndex(entry.getKey().partition());
- offsetData.computeIfAbsent(topic, key -> new ArrayList<>()).add(partition);
- }
- for (Map.Entry<String, List<OffsetCommitRequestPartition>> entry : offsetData.entrySet()) {
- OffsetCommitRequestTopic topic = new OffsetCommitRequestTopic()
- .setName(entry.getKey())
- .setPartitions(entry.getValue());
- topics.add(topic);
+ private void validateKeys(
+ Set<CoordinatorKey> groupIds
+ ) {
Review comment:
I think we put args on the same line unless the list is too big
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
Should REBALANCE_IN_PROGRESS and GROUP_AUTHORIZATION_FAILED be added to `groupsToRetry`?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
Review comment:
We can use `ApiResult.completed()`
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
Review comment:
We can use `ApiResult.unmapped()`
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
Review comment:
Is this a group-level error?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac merged pull request #11016:
URL: https://github.com/apache/kafka/pull/11016
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#issuecomment-882493385
@rajinisivaram Thanks for your comments. I have replied to them add/or addressed them.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac merged pull request #11016:
URL: https://github.com/apache/kafka/pull/11016
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r667478130
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/AlterConsumerGroupOffsetsResult.java
##########
@@ -48,23 +46,19 @@
public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
- this.future.whenComplete(new BiConsumer<Map<TopicPartition, Errors>, Throwable>() {
- @Override
- public void accept(final Map<TopicPartition, Errors> topicPartitions, final Throwable throwable) {
- if (throwable != null) {
- result.completeExceptionally(throwable);
- } else if (!topicPartitions.containsKey(partition)) {
- result.completeExceptionally(new IllegalArgumentException(
- "Alter offset for partition \"" + partition + "\" was not attempted"));
+ this.future.whenComplete((topicPartitions, throwable) -> {
Review comment:
Right.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/AlterConsumerGroupOffsetsResult.java
##########
@@ -75,22 +69,19 @@ public void accept(final Map<TopicPartition, Errors> topicPartitions, final Thro
* Return a future which succeeds if all the alter offsets succeed.
*/
public KafkaFuture<Void> all() {
- return this.future.thenApply(new BaseFunction<Map<TopicPartition, Errors>, Void>() {
- @Override
- public Void apply(final Map<TopicPartition, Errors> topicPartitionErrorsMap) {
- List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
- .stream()
- .filter(e -> e.getValue() != Errors.NONE)
- .map(Map.Entry::getKey)
- .collect(Collectors.toList());
- for (Errors error : topicPartitionErrorsMap.values()) {
- if (error != Errors.NONE) {
- throw error.exception(
- "Failed altering consumer group offsets for the following partitions: " + partitionsFailed);
- }
+ return this.future.thenApply(topicPartitionErrorsMap -> {
+ List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
Review comment:
Right.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672231624
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
Review 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
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review 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?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
https://issues.apache.org/jira/browse/KAFKA-13103
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rajinisivaram commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
rajinisivaram commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672241651
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
Yes, we can open a JIRA to do it later.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rajinisivaram commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
rajinisivaram commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672176498
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -73,29 +75,40 @@ public String apiName() {
return AdminApiFuture.forKeys(Collections.singleton(CoordinatorKey.byGroupId(groupId)));
}
- @Override
- public OffsetCommitRequest.Builder buildRequest(int coordinatorId, Set<CoordinatorKey> keys) {
- List<OffsetCommitRequestTopic> topics = new ArrayList<>();
- Map<String, List<OffsetCommitRequestPartition>> offsetData = new HashMap<>();
- for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
- String topic = entry.getKey().topic();
- OffsetAndMetadata oam = entry.getValue();
- OffsetCommitRequestPartition partition = new OffsetCommitRequestPartition()
- .setCommittedOffset(oam.offset())
- .setCommittedLeaderEpoch(oam.leaderEpoch().orElse(-1))
- .setCommittedMetadata(oam.metadata())
- .setPartitionIndex(entry.getKey().partition());
- offsetData.computeIfAbsent(topic, key -> new ArrayList<>()).add(partition);
- }
- for (Map.Entry<String, List<OffsetCommitRequestPartition>> entry : offsetData.entrySet()) {
- OffsetCommitRequestTopic topic = new OffsetCommitRequestTopic()
- .setName(entry.getKey())
- .setPartitions(entry.getValue());
- topics.add(topic);
+ private void validateKeys(
+ Set<CoordinatorKey> groupIds
+ ) {
Review comment:
I think we put args on the same line unless the list is too big
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
Should REBALANCE_IN_PROGRESS and GROUP_AUTHORIZATION_FAILED be added to `groupsToRetry`?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
Review comment:
We can use `ApiResult.completed()`
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
Review comment:
We can use `ApiResult.unmapped()`
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
Review comment:
Is this a group-level error?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
Yes, we can open a JIRA to do it later.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672231624
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
Review 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
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r667475174
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/AlterConsumerGroupOffsetsResult.java
##########
@@ -75,22 +69,19 @@ public void accept(final Map<TopicPartition, Errors> topicPartitions, final Thro
* Return a future which succeeds if all the alter offsets succeed.
*/
public KafkaFuture<Void> all() {
- return this.future.thenApply(new BaseFunction<Map<TopicPartition, Errors>, Void>() {
- @Override
- public Void apply(final Map<TopicPartition, Errors> topicPartitionErrorsMap) {
- List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
- .stream()
- .filter(e -> e.getValue() != Errors.NONE)
- .map(Map.Entry::getKey)
- .collect(Collectors.toList());
- for (Errors error : topicPartitionErrorsMap.values()) {
- if (error != Errors.NONE) {
- throw error.exception(
- "Failed altering consumer group offsets for the following partitions: " + partitionsFailed);
- }
+ return this.future.thenApply(topicPartitionErrorsMap -> {
+ List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
Review comment:
change to lambda expression?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/AlterConsumerGroupOffsetsResult.java
##########
@@ -48,23 +46,19 @@
public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
- this.future.whenComplete(new BiConsumer<Map<TopicPartition, Errors>, Throwable>() {
- @Override
- public void accept(final Map<TopicPartition, Errors> topicPartitions, final Throwable throwable) {
- if (throwable != null) {
- result.completeExceptionally(throwable);
- } else if (!topicPartitions.containsKey(partition)) {
- result.completeExceptionally(new IllegalArgumentException(
- "Alter offset for partition \"" + partition + "\" was not attempted"));
+ this.future.whenComplete((topicPartitions, throwable) -> {
Review comment:
This is just a change to lambda expression, no content change, right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672233510
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review 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?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#issuecomment-877771635
I am reviewing the other handlers which have been introduced in KIP-699. I will follow-up with other PRs if necessary.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#issuecomment-882608400
Failed tests are not related:
```
Build / ARM / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / ARM / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 16 and Scala 2.13 / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 16 and Scala 2.13 / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 8 and Scala 2.12 / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 8 and Scala 2.12 / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 8 and Scala 2.12 / kafka.server.epoch.LeaderEpochIntegrationTest.shouldAddCurrentLeaderEpochToMessagesAsTheyAreWrittenToLeader()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.clients.ClientUtilsTest.testResolveDnsLookup()
Build / JDK 16 and Scala 2.13 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
Build / JDK 16 and Scala 2.13 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672231624
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
Review 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
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review 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?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
https://issues.apache.org/jira/browse/KAFKA-13103
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rajinisivaram commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
rajinisivaram commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672241651
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
Yes, we can open a JIRA to do it later.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#issuecomment-882493385
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#issuecomment-882493385
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac merged pull request #11016:
URL: https://github.com/apache/kafka/pull/11016
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#discussion_r672252610
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##########
@@ -105,53 +118,96 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
+ final Map<TopicPartition, Errors> partitionResults = new HashMap<>();
- Map<TopicPartition, Errors> partitions = new HashMap<>();
for (OffsetCommitResponseTopic topic : response.data().topics()) {
for (OffsetCommitResponsePartition partition : topic.partitions()) {
- TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
+ TopicPartition topicPartition = new TopicPartition(topic.name(), partition.partitionIndex());
Errors error = Errors.forCode(partition.errorCode());
+
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleError(
+ groupId,
+ topicPartition,
+ error,
+ partitionResults,
+ groupsToUnmap,
+ groupsToRetry
+ );
} else {
- partitions.put(tp, error);
+ partitionResults.put(topicPartition, error);
}
}
}
- if (failed.isEmpty() && unmapped.isEmpty())
- completed.put(groupId, partitions);
- return new ApiResult<>(completed, failed, unmapped);
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ Collections.singletonMap(groupId, partitionResults),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ } else {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
+ }
}
private void handleError(
CoordinatorKey groupId,
+ TopicPartition topicPartition,
Errors error,
- Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Map<TopicPartition, Errors> partitionResults,
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- log.error("Received authorization failure for group {} in `OffsetCommit` response", groupId,
- error.exception());
- failed.put(groupId, error.exception());
- break;
+ // 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 {} returned error {}. Will retry", groupId, error);
- unmapped.add(groupId);
+ 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:
+ case GROUP_AUTHORIZATION_FAILED:
+ log.debug("OffsetCommit request for group id {} failed due to error {}.",
+ groupId.idValue, error);
+ partitionResults.put(topicPartition, error);
Review comment:
https://issues.apache.org/jira/browse/KAFKA-13103
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11016: KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly.
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11016:
URL: https://github.com/apache/kafka/pull/11016#issuecomment-882610943
Merged to trunk and 3.0.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org