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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
