jeffkbkim commented on code in PR #16057:
URL: https://github.com/apache/kafka/pull/16057#discussion_r1613555271
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4424,14 +4441,113 @@ private ConsumerGroupMember
validateConsumerGroupMember(
* @param context The request context.
* @param request The actual LeaveGroup request.
*
+ * @return The LeaveGroup response and the records to append.
+ */
+ public CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord>
classicGroupLeave(
+ RequestContext context,
+ LeaveGroupRequestData request
+ ) throws UnknownMemberIdException, GroupIdNotFoundException {
+ Group group = groups.get(request.groupId(), Long.MAX_VALUE);
+
+ if (group == null) {
+ throw new UnknownMemberIdException(String.format("Group %s not
found.", request.groupId()));
+ }
Review Comment:
i'm guessing we don't use
```
Group group = group(request.memberId());
```
as we throw unknown member id for the old coordinator, is this correct? it
seems very counterintuitive unfortunately
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4424,14 +4441,113 @@ private ConsumerGroupMember
validateConsumerGroupMember(
* @param context The request context.
* @param request The actual LeaveGroup request.
*
+ * @return The LeaveGroup response and the records to append.
+ */
+ public CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord>
classicGroupLeave(
+ RequestContext context,
+ LeaveGroupRequestData request
+ ) throws UnknownMemberIdException, GroupIdNotFoundException {
+ Group group = groups.get(request.groupId(), Long.MAX_VALUE);
+
+ if (group == null) {
+ throw new UnknownMemberIdException(String.format("Group %s not
found.", request.groupId()));
+ }
+
+ if (group.type() == CLASSIC) {
+ return classicGroupLeaveToClassicGroup((ClassicGroup) group,
context, request);
+ } else {
+ return classicGroupLeaveToConsumerGroup((ConsumerGroup) group,
context, request);
+ }
+ }
+
+ /**
+ * Handle a classic LeaveGroupRequest to a ConsumerGroup.
+ *
+ * @param group The ConsumerGroup.
+ * @param context The request context.
+ * @param request The actual LeaveGroup request.
+ *
+ * @return The LeaveGroup response and the records to append.
+ */
+ private CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord>
classicGroupLeaveToConsumerGroup(
+ ConsumerGroup group,
+ RequestContext context,
+ LeaveGroupRequestData request
+ ) throws UnknownMemberIdException, GroupIdNotFoundException {
+ List<MemberResponse> memberResponses = new ArrayList<>();
+ List<CoordinatorRecord> records = new ArrayList<>();
+ boolean hasValidLeaveGroupMember = false;
+
+ for (MemberIdentity memberIdentity: request.members()) {
+ String memberId = memberIdentity.memberId();
+ String instanceId = memberIdentity.groupInstanceId();
+ String reason = memberIdentity.reason() != null ?
memberIdentity.reason() : "not provided";
+
+ ConsumerGroupMember member;
+ try {
+ if (instanceId == null) {
+ member = group.getOrMaybeCreateMember(memberId, false);
+ throwIfMemberDoesNotUseClassicProtocol(member);
+
+ log.info("[Group {}] Static Member {} has left group " +
Review Comment:
nit: this should not be "Static Member"
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4424,14 +4441,113 @@ private ConsumerGroupMember
validateConsumerGroupMember(
* @param context The request context.
* @param request The actual LeaveGroup request.
*
+ * @return The LeaveGroup response and the records to append.
+ */
+ public CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord>
classicGroupLeave(
+ RequestContext context,
+ LeaveGroupRequestData request
+ ) throws UnknownMemberIdException, GroupIdNotFoundException {
+ Group group = groups.get(request.groupId(), Long.MAX_VALUE);
+
+ if (group == null) {
+ throw new UnknownMemberIdException(String.format("Group %s not
found.", request.groupId()));
+ }
+
+ if (group.type() == CLASSIC) {
+ return classicGroupLeaveToClassicGroup((ClassicGroup) group,
context, request);
+ } else {
+ return classicGroupLeaveToConsumerGroup((ConsumerGroup) group,
context, request);
+ }
+ }
+
+ /**
+ * Handle a classic LeaveGroupRequest to a ConsumerGroup.
+ *
+ * @param group The ConsumerGroup.
+ * @param context The request context.
+ * @param request The actual LeaveGroup request.
+ *
+ * @return The LeaveGroup response and the records to append.
+ */
+ private CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord>
classicGroupLeaveToConsumerGroup(
+ ConsumerGroup group,
+ RequestContext context,
+ LeaveGroupRequestData request
+ ) throws UnknownMemberIdException, GroupIdNotFoundException {
+ List<MemberResponse> memberResponses = new ArrayList<>();
+ List<CoordinatorRecord> records = new ArrayList<>();
+ boolean hasValidLeaveGroupMember = false;
+
+ for (MemberIdentity memberIdentity: request.members()) {
+ String memberId = memberIdentity.memberId();
+ String instanceId = memberIdentity.groupInstanceId();
+ String reason = memberIdentity.reason() != null ?
memberIdentity.reason() : "not provided";
+
+ ConsumerGroupMember member;
+ try {
+ if (instanceId == null) {
+ member = group.getOrMaybeCreateMember(memberId, false);
+ throwIfMemberDoesNotUseClassicProtocol(member);
+
+ log.info("[Group {}] Static Member {} has left group " +
+ "through explicit `LeaveGroup` request; client
reason: {}",
+ group.groupId(), memberId, reason);
+ } else {
+ member = group.staticMember(instanceId);
+ throwIfStaticMemberIsUnknown(member, memberId);
+ // The LeaveGroup API allows administrative removal of
members by GroupInstanceId
+ // in which case we expect the MemberId to be undefined.
+ if (!UNKNOWN_MEMBER_ID.equals(memberId)) {
+ throwIfInstanceIdIsFenced(member, group.groupId(),
memberId, instanceId);
+ }
+ throwIfMemberDoesNotUseClassicProtocol(member);
+
+ log.info("[Group {}] Static Member {} with instance id {}
has left group " +
+ "through explicit `LeaveGroup` request; client
reason: {}",
+ group.groupId(), memberId, instanceId, reason);
+ }
+
+
records.addAll(removeMemberAndMaybeUpdateSubscriptionMetadata(group, member));
Review Comment:
this is because we want to invoke `group.computeSubscriptionMetadata()` once
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
