KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly. by dajac · Pull Request #11016 · apache/kafka

dajac · 2021-07-10T16:47:23Z

While reviewing #10973, I have noticed that AlterConsumerGroupOffsetsHandler does not handle partition errors correctly. The issue is that any partition error fails the entire future instead of being passed as an error for its corresponding partition. KafkaAdminClientTest#testOffsetCommitWithMultipleErrors reproduces the bug.

The regression was introduced by KIP-699.

Context: #10973 (comment).

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…tion errors correctly.

dajac · 2021-07-11T09:51:05Z

I am reviewing the other handlers which have been introduced in KIP-699. I will follow-up with other PRs if necessary.

showuon

@dajac , thanks for your nice catch and quick fix! I confirmed that the handleResponse in AlterConsumerGroupOffsetsHandler has the same logic as before. That is, we'll always return each partition result (even with error), except coordinator related errors (that needs retry or unmap).

Here is the previous code for other reviewer's reference:

 void handleResponse(AbstractResponse abstractResponse) {
      final OffsetCommitResponse response = (OffsetCommitResponse) abstractResponse;
  
      Map<Errors, Integer> errorCounts = response.errorCounts();
      // 1) If coordinator changed since we fetched it, retry
      // 2) If there is a coordinator error, retry
      if (ConsumerGroupOperationContext.hasCoordinatorMoved(errorCounts) ||
          ConsumerGroupOperationContext.shouldRefreshCoordinator(errorCounts)) {
          Call call = getAlterConsumerGroupOffsetsCall(context, offsets);
          rescheduleFindCoordinatorTask(context, () -> call, this);
          return;
      }
  
      final 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());
              Errors error = Errors.forCode(partition.errorCode());
              partitions.put(tp, error);
          }
      }
      context.future().complete(partitions);
  }

Thank you.

showuon · 2021-07-11T12:30:32Z

-                } else if (!topicPartitions.containsKey(partition)) {
-                    result.completeExceptionally(new IllegalArgumentException(
-                        "Alter offset for partition \"" + partition + "\" was not attempted"));
+        this.future.whenComplete((topicPartitions, throwable) -> {


This is just a change to lambda expression, no content change, right?

showuon · 2021-07-11T12:35:26Z

-                            "Failed altering consumer group offsets for the following partitions: " + partitionsFailed);
-                    }
+        return this.future.thenApply(topicPartitionErrorsMap ->  {
+            List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()


change to lambda expression?

rajinisivaram

@dajac Thanks for the PR. Left a few minor comments/questions, apart from that LGTM

rajinisivaram · 2021-07-19T10:23:23Z

-            topics.add(topic);
+    private void validateKeys(
+        Set<CoordinatorKey> groupIds
+    ) {


I think we put args on the same line unless the list is too big

rajinisivaram · 2021-07-19T10:42:08Z

+            case GROUP_AUTHORIZATION_FAILED:
+                log.debug("OffsetCommit request for group id {} failed due to error {}.",
+                    groupId.idValue, error);
+                partitionResults.put(topicPartition, error);


Should REBALANCE_IN_PROGRESS and GROUP_AUTHORIZATION_FAILED be added to groupsToRetry?

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?

Yes, we can open a JIRA to do it later.

https://issues.apache.org/jira/browse/KAFKA-13103

rajinisivaram · 2021-07-19T10:45:10Z

+            // Group level errors.
+            case INVALID_GROUP_ID:
+            case REBALANCE_IN_PROGRESS:
+            case INVALID_COMMIT_OFFSET_SIZE:


Is this a group-level error?

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

rajinisivaram · 2021-07-19T10:47:19Z


-        return new ApiResult<>(completed, failed, unmapped);
+        if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+            return new ApiResult<>(


We can use ApiResult.completed()

rajinisivaram · 2021-07-19T10:49:29Z

+                Collections.emptyList()
+            );
+        } else {
+            return new ApiResult<>(


We can use ApiResult.unmapped()

dajac · 2021-07-19T12:05:28Z

@rajinisivaram Thanks for your comments. I have replied to them add/or addressed them.

rajinisivaram

@dajac Thanks for the updates, LGTM

dajac · 2021-07-19T14:46:00Z

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()

…on errors correctly. (#11016) This patch updates `AlterConsumerGroupOffsetsHandler` to handle partition errors correctly. The issue is that any partition error fails the entire future instead of being passed as an error for its corresponding partition. Reviewers: Luke Chen <showuon@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

dajac · 2021-07-19T14:49:21Z

Merged to trunk and 3.0.

…on errors correctly. (apache#11016) This patch updates `AlterConsumerGroupOffsetsHandler` to handle partition errors correctly. The issue is that any partition error fails the entire future instead of being passed as an error for its corresponding partition. Reviewers: Luke Chen <showuon@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle parti…

014f2c6

…tion errors correctly.

dajac force-pushed the minor-kip-699 branch from b587250 to 014f2c6 Compare July 11, 2021 09:41

dajac changed the title ~~[WIP] Update AlterConsumerGroupOffsetsHandler to handle errors correctly.~~ KAFKA-13058; AlterConsumerGroupOffsetsHandler does not handle partition errors correctly. Jul 11, 2021

dajac mentioned this pull request Jul 11, 2021

KAFKA-13033: COORDINATOR_NOT_AVAILABLE should be unmapped #10973

Closed

3 tasks

dajac marked this pull request as ready for review July 11, 2021 09:49

dajac requested review from mimaison and rajinisivaram July 11, 2021 09:49

showuon approved these changes Jul 11, 2021

View reviewed changes

showuon mentioned this pull request Jul 12, 2021

KAFKA-13059: Make DeleteConsumerGroupOffsetsHandler unmap for COORDINATOR_NOT_AVAILABLE error and fix issue #11019

Merged

3 tasks

rajinisivaram reviewed Jul 19, 2021

View reviewed changes

Address comment

b8432aa

rajinisivaram approved these changes Jul 19, 2021

View reviewed changes

dajac merged commit f7790a0 into apache:trunk Jul 19, 2021

dajac deleted the minor-kip-699 branch July 19, 2021 14:49

Conversation

dajac commented Jul 10, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

dajac commented Jul 11, 2021

Uh oh!

showuon left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rajinisivaram left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac commented Jul 19, 2021

Uh oh!

rajinisivaram left a comment

Choose a reason for hiding this comment

Uh oh!

dajac commented Jul 19, 2021

Uh oh!

dajac commented Jul 19, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dajac commented Jul 10, 2021 •

edited

Loading

showuon left a comment •

edited

Loading