KAFKA-13063: Make DescribeConsumerGroupsHandler unmap for COORDINATOR_NOT_AVAILABLE error by showuon · Pull Request #11022 · apache/kafka

showuon · 2021-07-12T10:12:42Z

refactor DescribeConsumerGroupsHandler and tests. Also, put COORDINATOR_NOT_AVAILABLE as unmap retry.

the old handleResponse for DescribeConsumerGroups request:

void handleResponse(AbstractResponse abstractResponse) {
      final DescribeGroupsResponse response = (DescribeGroupsResponse) abstractResponse;

      List<DescribedGroup> describedGroups = response.data().groups();
      if (describedGroups.isEmpty()) {
          context.future().completeExceptionally(
                  new InvalidGroupIdException("No consumer group found for GroupId: " + context.groupId()));
          return;
      }

      if (describedGroups.size() > 1 ||
              !describedGroups.get(0).groupId().equals(context.groupId())) {
          String ids = Arrays.toString(describedGroups.stream().map(DescribedGroup::groupId).toArray());
          context.future().completeExceptionally(new InvalidGroupIdException(
                  "DescribeConsumerGroup request for GroupId: " + context.groupId() + " returned " + ids));
          return;
      }

      final DescribedGroup describedGroup = describedGroups.get(0);

      // If coordinator changed since we fetched it, retry
      if (ConsumerGroupOperationContext.hasCoordinatorMoved(response)) {
          Call call = getDescribeConsumerGroupsCall(context);
          rescheduleFindCoordinatorTask(context, () -> call, this);
          return;
      }

      final Errors groupError = Errors.forCode(describedGroup.errorCode());
      if (handleGroupRequestError(groupError, context.future()))
          return;

      final String protocolType = describedGroup.protocolType();
      if (protocolType.equals(ConsumerProtocol.PROTOCOL_TYPE) || protocolType.isEmpty()) {
          final List<DescribedGroupMember> members = describedGroup.members();
          final List<MemberDescription> memberDescriptions = new ArrayList<>(members.size());
          final Set<AclOperation> authorizedOperations = validAclOperations(describedGroup.authorizedOperations());
          for (DescribedGroupMember groupMember : members) {
              Set<TopicPartition> partitions = Collections.emptySet();
              if (groupMember.memberAssignment().length > 0) {
                  final Assignment assignment = ConsumerProtocol.
                      deserializeAssignment(ByteBuffer.wrap(groupMember.memberAssignment()));
                  partitions = new HashSet<>(assignment.partitions());
              }
              final MemberDescription memberDescription = new MemberDescription(
                      groupMember.memberId(),
                      Optional.ofNullable(groupMember.groupInstanceId()),
                      groupMember.clientId(),
                      groupMember.clientHost(),
                      new MemberAssignment(partitions));
              memberDescriptions.add(memberDescription);
          }
          final ConsumerGroupDescription consumerGroupDescription =
              new ConsumerGroupDescription(context.groupId(), protocolType.isEmpty(),
                  memberDescriptions,
                  describedGroup.protocolData(),
                  ConsumerGroupState.parse(describedGroup.groupState()),
                  context.node().get(),
                  authorizedOperations);
          context.future().complete(consumerGroupDescription);
      } else {
          context.future().completeExceptionally(new IllegalArgumentException(
              String.format("GroupId %s is not a consumer group (%s).",
                  context.groupId(), protocolType)));
      }
  }

Committer Checklist (excluded from commit message)

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

dajac · 2021-07-13T15:03:45Z

+    private void validateGroupsNotEmpty(List<DescribedGroup> describedGroups) {
+        if (describedGroups.isEmpty()) {
+            throw new InvalidGroupIdException("No consumer group found");
+        }
+    }


What's the purpose of this check? I am not sure where this InvalidGroupIdException thrown here will get to.

Before KIP-699, there's the check:

if (describedGroups.isEmpty()) { context.future().completeExceptionally( new InvalidGroupIdException("No consumer group found for GroupId: " + context.groupId())); return; }

But, you're right, the exception thrown will not return back to requester. Also, because there's no group found, we cannot have a failed key return because we need a group in key.

Removed it since it should never happen.

dajac · 2021-07-13T15:04:01Z

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


It seems incorrect here as well.

You're right! We accept multiple groups. Updated. Thanks.

dajac

@showuon Thank for the update. I left a few more comments.

dajac · 2021-07-14T17:04:18Z

            Errors error = Errors.forCode(describedGroup.errorCode());
            if (error != Errors.NONE) {
-                handleError(groupIdKey, error, failed, unmapped);
+                handleError(groupIdKey, error, failed, groupsToUnmap, groupsToRetry);


groupsToRetry to retry is not really necessary in this case. We don't even use it later. Could we remove it?

Good catch! Updated.

dajac · 2021-07-14T17:04:42Z

        }
-        return new ApiResult<>(completed, failed, unmapped);
+
+        return new ApiResult<>(completed, failed,  new ArrayList<>(groupsToUnmap));


nit: There is an extra space before new.

dajac · 2021-07-14T17:05:43Z

+                final String unexpectedErrorMsg =
+                    String.format("`DescribeGroups` request for group id %s failed due to error %s", groupId.idValue, error);
+                log.error(unexpectedErrorMsg);
+                failed.put(groupId, error.exception(unexpectedErrorMsg));


We don't provide the error message in any other case. Should we remove this one for the time being? I think that it is a good idea but only if we do it across the board.

Agree. I just follow the previous behavior. Remove the error message. Thanks.

dajac · 2021-07-14T17:05:54Z

    }

-}
+}


nit: Could we revert this back?

showuon · 2021-07-15T07:50:12Z

@dajac , I addressed your comments in this PR and all other 4 PRs. And also update the PR title and description accordingly. Please help take a look when available. Thank you.

dajac

LGTM

dajac · 2021-07-15T12:22:43Z

Failures are not related:

Build / JDK 8 and Scala 2.12 / testUnauthorizedDeleteTopicsWithDescribe() – kafka.api.AuthorizerIntegrationTest
1m 7s
Build / JDK 8 and Scala 2.12 / testCloseDuringRebalance() – kafka.api.ConsumerBounceTest
18s
Build / JDK 8 and Scala 2.12 / testListReassignmentsDoesNotShowNonReassigningPartitions() – kafka.api.PlaintextAdminIntegrationTest
5s
Build / JDK 8 and Scala 2.12 / testSendOffsetsToTransactionTimeout() – kafka.api.TransactionsTest

…_NOT_AVAILABLE error (#11022) This patch improve the error handling in `DescribeConsumerGroupsHandler` and ensure that `COORDINATOR_NOT_AVAILABLE` is unmapped in order to look up the coordinator again. Reviewers: David Jacot <djacot@confluent.io>

dajac · 2021-07-15T12:26:37Z

Merged to trunk and to 3.0. cc @kkonstantine

…_NOT_AVAILABLE error (apache#11022) This patch improve the error handling in `DescribeConsumerGroupsHandler` and ensure that `COORDINATOR_NOT_AVAILABLE` is unmapped in order to look up the coordinator again. Reviewers: David Jacot <djacot@confluent.io>

KAFKA-13063: refactor DescribeConsumerGroupsHandler and tests

3038544

showuon mentioned this pull request Jul 13, 2021

KAFKA-13033: COORDINATOR_NOT_AVAILABLE should be unmapped #10973

Closed

3 tasks

dajac reviewed Jul 13, 2021

View reviewed changes

showuon added 2 commits July 14, 2021 15:28

KAFKA-13063: refactor codes

e9a1868

Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13063

60afa6c

showuon force-pushed the KAFKA-13063 branch from f7914b0 to 60afa6c Compare July 14, 2021 07:29

KAFKA-13063: refactor

816bc07

dajac reviewed Jul 14, 2021

View reviewed changes

KAFKA-13063: revert find coordinator response change

07ef9bf

showuon force-pushed the KAFKA-13063 branch from 28702d5 to 07ef9bf Compare July 15, 2021 03:53

showuon changed the title ~~KAFKA-13063: refactor DescribeConsumerGroupsHandler and tests~~ KAFKA-13063: Make DescribeConsumerGroupsHandler unmap for COORDINATOR_NOT_AVAILABLE error Jul 15, 2021

dajac approved these changes Jul 15, 2021

View reviewed changes

dajac merged commit df398aa into apache:trunk Jul 15, 2021

                   }
-              }
+              }

                
                    No newline at end of file

Conversation

showuon commented Jul 12, 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!

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 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!

Uh oh!

showuon commented Jul 15, 2021

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

dajac commented Jul 15, 2021

Uh oh!

dajac commented Jul 15, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

showuon commented Jul 12, 2021 •

edited

Loading