jeffkbkim commented on code in PR #13870: URL: https://github.com/apache/kafka/pull/13870#discussion_r1235752587
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -874,4 +1045,1265 @@ public void replay( consumerGroup.updateMember(newMember); } } + + // Below stores all methods to handle generic group APIs. + + /** + * Handle a JoinGroupRequest. + * + * @param context The request context. + * @param request The actual JoinGroup request. + * + * @return The result that contains records to append if the join group phase completes. + */ + public CoordinatorResult<CompletableFuture<Errors>, Record> genericGroupJoin( + RequestContext context, + JoinGroupRequestData request, + CompletableFuture<JoinGroupResponseData> responseFuture + ) { + CoordinatorResult<CompletableFuture<Errors>, Record> result = EMPTY_RESULT; + String groupId = request.groupId(); + String memberId = request.memberId(); + int sessionTimeoutMs = request.sessionTimeoutMs(); + + if (sessionTimeoutMs < groupMinSessionTimeoutMs || + sessionTimeoutMs > groupMaxSessionTimeoutMs + ) { + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(memberId) + .setErrorCode(Errors.INVALID_SESSION_TIMEOUT.code()) + ); + } else { + boolean isUnknownMember = memberId.equals(UNKNOWN_MEMBER_ID); + // Group is created if it does not exist and the member id is UNKNOWN. if member + // is specified but group does not exist, request is rejected with GROUP_ID_NOT_FOUND + GenericGroup group; + try { + group = (GenericGroup) getOrMaybeCreateGroup(groupId, GENERIC, isUnknownMember); + } catch (Throwable t) { + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(memberId) + .setErrorCode(Errors.forException(t).code()) + ); + return EMPTY_RESULT; + } + + String joinReason = request.reason(); + if (joinReason == null || joinReason.isEmpty()) { + joinReason = "not provided"; + } + + if (!acceptJoiningMember(group, memberId)) { + group.remove(memberId); + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(UNKNOWN_MEMBER_ID) + .setErrorCode(Errors.GROUP_MAX_SIZE_REACHED.code()) + ); + + } else if (isUnknownMember) { + result = genericGroupJoinNewMember( + context, + request, + group, + joinReason, + responseFuture + ); + } else { + result = genericGroupJoinExistingMember( + context, + request, + group, + joinReason, + responseFuture + ); + } + + // Attempt to complete join group phase. We do not complete + // the join group phase if this is the initial rebalance. + if (group.isInState(PREPARING_REBALANCE) && + group.hasAllMembersJoined() && + group.generationId() != 0 + ) { + completeGenericGroupJoin(group); + } + } + + return result; + } + + /** + * Handle a new member generic group join. + * + * @param context The request context. + * @param request The join group request. + * @param group The group to add the member. + * @param joinReason The client reason for the join request. + * @param responseFuture The response future to complete. + * + * @return The coordinator result that will be appended to the log. + */ + private CoordinatorResult<CompletableFuture<Errors>, Record> genericGroupJoinNewMember( + RequestContext context, + JoinGroupRequestData request, + GenericGroup group, + String joinReason, + CompletableFuture<JoinGroupResponseData> responseFuture + ) { + List<Protocol> protocols = new ArrayList<>(); + request.protocols().forEach(protocol -> protocols.add(new Protocol(protocol.name(), protocol.metadata()))); + if (group.isInState(DEAD)) { + // if the group is marked as dead, it means some other thread has just removed the group + // from the coordinator metadata; it is likely that the group has migrated to some other + // coordinator OR the group is in a transient unstable phase. Let the member retry + // finding the correct coordinator and rejoin. + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(UNKNOWN_MEMBER_ID) + .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()) + ); + } else if (!group.supportsProtocols(request.protocolType(), GenericGroupMember.plainProtocolSet(protocols))) { + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(UNKNOWN_MEMBER_ID) + .setErrorCode(Errors.INCONSISTENT_GROUP_PROTOCOL.code()) + ); + } else { + Optional<String> groupInstanceId = Optional.ofNullable(request.groupInstanceId()); + String newMemberId = group.generateMemberId(context.clientId(), groupInstanceId); + + if (groupInstanceId.isPresent()) { + return genericGroupJoinNewStaticMember( + context, + request, + group, + newMemberId, + joinReason, + responseFuture + ); + } else { + return genericGroupJoinNewDynamicMember( + context, + request, + group, + newMemberId, + joinReason, + responseFuture + ); + } + } + + return EMPTY_RESULT; + } + + /** + * Handle new static member join. If there was an existing member id for the group instance id, + * replace that member. Otherwise, add the member and rebalance. + * + * @param context The request context. + * @param request The join group request. + * @param group The group to add the member. + * @param newMemberId The newly generated member id. + * @param joinReason The client reason for the join request. + * @param responseFuture The response future to complete. + * + * @return The coordinator result that will be appended to the log. + */ + private CoordinatorResult<CompletableFuture<Errors>, Record> genericGroupJoinNewStaticMember( + RequestContext context, + JoinGroupRequestData request, + GenericGroup group, + String newMemberId, + String joinReason, + CompletableFuture<JoinGroupResponseData> responseFuture + ) { + String groupInstanceId = request.groupInstanceId(); + String existingMemberId = group.staticMemberId(groupInstanceId); + if (existingMemberId != null) { + log.info("Static member with groupInstanceId={} and unknown member id joins " + + "group {} in {} state. Replacing previously mapped member {} with this groupInstanceId.", + groupInstanceId, group.groupId(), group.currentState(), existingMemberId); + + return updateStaticMemberAndRebalance( + context, + request, + group, + existingMemberId, + newMemberId, + joinReason, + responseFuture + ); + + } else { + log.info("Static member with groupInstanceId={} and unknown member id joins " + + "group {} in {} state. Created a new member id {} for this member and added to the group.", + groupInstanceId, group.groupId(), group.currentState(), newMemberId); + + return addMemberAndRebalance(context, request, group, newMemberId, joinReason, responseFuture); + } + } + + /** + * Handle a new dynamic member join. If the member id field is required, the group metadata manager + * will add the new member id to the pending members and respond with MEMBER_ID_REQUIRED along with + * the new member id for the client to join with. + * + * Otherwise, add the new member to the group and rebalance. + * + * @param context The request context. + * @param request The join group request. + * @param group The group to add the member. + * @param newMemberId The newly generated member id. + * @param joinReason The client reason for the join request. + * @param responseFuture The response future to complete. + * + * @return The coordinator result that will be appended to the log. + */ + private CoordinatorResult<CompletableFuture<Errors>, Record> genericGroupJoinNewDynamicMember( + RequestContext context, + JoinGroupRequestData request, + GenericGroup group, + String newMemberId, + String joinReason, + CompletableFuture<JoinGroupResponseData> responseFuture + ) { + boolean requireKnownMemberId = context.apiVersion() >= 4 + && (request.groupInstanceId() == null || request.groupInstanceId().isEmpty()); + + if (requireKnownMemberId) { + // If member id required, register the member in the pending member list and send + // back a response to call for another join group request with allocated member id. + log.info("Dynamic member with unknown member id joins group {} in {} state. " + + "Created a new member id {} and requesting the member to rejoin with this id.", + group.groupId(), group.currentState(), newMemberId); + + group.addPendingMember(newMemberId); + String heartbeatKey = heartbeatKey(group.groupId(), newMemberId); + + timer.schedule( + heartbeatKey, + request.sessionTimeoutMs(), + TimeUnit.MILLISECONDS, + () -> expireGenericGroupMemberHeartbeat(group, newMemberId) + ); + + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(newMemberId) + .setErrorCode(Errors.MEMBER_ID_REQUIRED.code()) + ); + + } else { + log.info("Dynamic member with unknown member id joins group {} in state {}. " + + "Created a new member id {} and added the member to the group.", + group.groupId(), group.currentState(), newMemberId); + + return addMemberAndRebalance(context, request, group, newMemberId, joinReason, responseFuture); + } + + return EMPTY_RESULT; + } + + /** + * Handle a join group request for an existing member. + * + * @param context The request context. + * @param request The join group request. + * @param group The group to add the member. + * @param joinReason The client reason for the join request. + * @param responseFuture The response future to complete. + * + * @return The coordinator result that will be appended to the log. + */ + private CoordinatorResult<CompletableFuture<Errors>, Record> genericGroupJoinExistingMember( + RequestContext context, + JoinGroupRequestData request, + GenericGroup group, + String joinReason, + CompletableFuture<JoinGroupResponseData> responseFuture + ) { + List<Protocol> protocols = request.protocols().stream() + .map(protocol -> new Protocol(protocol.name(), protocol.metadata())).collect(Collectors.toList()); + + String memberId = request.memberId(); + String groupInstanceId = request.groupInstanceId(); + + if (group.isInState(DEAD)) { + // If the group is marked as dead, it means the group was recently removed the group + // from the coordinator metadata; it is likely that the group has migrated to some other + // coordinator OR the group is in a transient unstable phase. Let the member retry + // finding the correct coordinator and rejoin. + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(memberId) + .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()) + ); + } else if (!group.supportsProtocols( + request.protocolType(), + GenericGroupMember.plainProtocolSet(protocols)) + ) { + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(memberId) + .setErrorCode(Errors.INCONSISTENT_GROUP_PROTOCOL.code()) + ); + } else if (group.isPendingMember(memberId)) { + // A rejoining pending member will be accepted. Note that pending member cannot be a static member. + if (groupInstanceId != null) { + throw new IllegalStateException("Received unexpected JoinGroup with groupInstanceId=" + + groupInstanceId + " for pending member with memberId=" + memberId); + } + + log.debug("Pending dynamic member with id {} joins group {} in {} state. Adding to the group now.", + memberId, group.groupId(), group.currentState()); + + return addMemberAndRebalance( + context, + request, + group, + memberId, + joinReason, + responseFuture + ); + } else { + Optional<Errors> memberError = validateExistingMember( + group, + memberId, + groupInstanceId, + "join-group" + ); + + if (memberError.isPresent()) { + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(memberId) + .setErrorCode(memberError.get().code()) + ); + } else { + GenericGroupMember member = group.member(memberId); + if (group.isInState(PREPARING_REBALANCE)) { + return updateMemberAndRebalance( + request, + group, + member, + "Member " + member.memberId() + " is joining group during " + + group.stateAsString() + "; client reason: " + joinReason, + responseFuture + ); + } else if (group.isInState(COMPLETING_REBALANCE)) { + if (member.matches(protocols)) { + // Member is joining with the same metadata (which could be because it failed to + // receive the initial JoinGroup response), so just return current group information + // for the current generation. + List<JoinGroupResponseData.JoinGroupResponseMember> members = Collections.emptyList(); + if (group.isLeader(memberId)) { + members = group.currentGenericGroupMembers(); + } + responseFuture.complete(new JoinGroupResponseData() + .setMembers(members) + .setMemberId(memberId) + .setGenerationId(group.generationId()) + .setProtocolName(group.protocolName().orElse(null)) + .setProtocolType(group.protocolType().orElse(null)) + .setLeader(group.leaderOrNull()) + .setSkipAssignment(false) + .setErrorCode(Errors.NONE.code()) + ); + } else { + // Member has changed metadata, so force a rebalance + return updateMemberAndRebalance( + request, + group, + member, + "Updating metadata for member " + memberId + " during " + + group.stateAsString() + "; client reason: " + joinReason, + responseFuture + ); + } + } else if (group.isInState(STABLE)) { + if (group.isLeader(memberId)) { + // Force a rebalance if the leader sends JoinGroup; + // This allows the leader to trigger rebalances for changes affecting assignment + // which do not affect the member metadata (such as topic metadata changes for the consumer) + return updateMemberAndRebalance( + request, + group, + member, + "Leader " + memberId + " re-joining group during " + + group.stateAsString() + "; client reason: " + joinReason, + responseFuture + ); + } else if (!member.matches(protocols)) { + return updateMemberAndRebalance( + request, + group, + member, + "Updating metadata for member " + memberId + " during " + + group.stateAsString() + "; client reason: " + joinReason, + responseFuture + ); + } else { + // For followers with no actual change to their metadata, just return group information + // for the current generation which will allow them to issue SyncGroup. + responseFuture.complete(new JoinGroupResponseData() + .setMembers(Collections.emptyList()) + .setMemberId(memberId) + .setGenerationId(group.generationId()) + .setProtocolName(group.protocolName().orElse(null)) + .setProtocolType(group.protocolType().orElse(null)) + .setLeader(group.leaderOrNull()) + .setSkipAssignment(false) + .setErrorCode(Errors.NONE.code()) + ); + } + } else { + // Group reached unexpected (Empty) state. Let the joining member reset their generation and rejoin. + log.warn("Attempt to add rejoining member {} of group {} in unexpected group state {}", + memberId, group.groupId(), group.stateAsString()); + + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(memberId) + .setErrorCode(Errors.UNKNOWN_MEMBER_ID.code()) + ); + } + } + } + + return EMPTY_RESULT; + } + + /** + * Complete the join group phase. Remove all dynamic members that have not rejoined + * during this stage and proceed with the next generation for this group. The generation id + * is incremented and the group transitions to CompletingRebalance state if there is at least + * one member. + * + * If the group is in Empty state, append a new group metadata record to the log. Otherwise, + * complete all members' join group response futures and wait for sync requests from members. + * + * @param group The group that is completing the join group phase. + * + * @return The coordinator result that will be appended to the log. + */ + private CoordinatorResult<CompletableFuture<Errors>, Record> completeGenericGroupJoin( + GenericGroup group + ) { + timer.cancel(joinKey(group.groupId())); + String groupId = group.groupId(); + + Map<String, GenericGroupMember> notYetRejoinedDynamicMembers = + group.notYetRejoinedMembers().entrySet().stream() + .filter(entry -> !entry.getValue().isStaticMember()) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + + if (!notYetRejoinedDynamicMembers.isEmpty()) { + notYetRejoinedDynamicMembers.values().forEach(failedMember -> { + group.remove(failedMember.memberId()); + timer.cancel(heartbeatKey(group.groupId(), failedMember.memberId())); + }); + + log.info("Group {} removed dynamic members who haven't joined: {}", + groupId, notYetRejoinedDynamicMembers.keySet()); + } + + if (group.isInState(DEAD)) { + log.info("Group {} is dead, skipping rebalance stage.", groupId); + } else if (!group.maybeElectNewJoinedLeader() && !group.allMembers().isEmpty()) { + // If all members are not rejoining, we will postpone the completion + // of rebalance preparing stage, and send out another delayed operation + // until session timeout removes all the non-responsive members. + log.error("Group {} could not complete rebalance because no members rejoined.", groupId); + + return tryCompleteJoinElseSchedule(group); + } else { + group.initNextGeneration(); + if (group.isInState(EMPTY)) { + log.info("Group {} with generation {} is now empty ({}-{})", + groupId, group.generationId(), topicPartition.topic(), topicPartition.partition()); + + CompletableFuture<Errors> appendFuture = new CompletableFuture<>(); + appendFuture.whenComplete((error, t) -> { + if (t != null || error != Errors.NONE) { + // We failed to write the empty group metadata. If the broker fails before another rebalance, + // the previous generation written to the log will become active again (and most likely timeout). + // This should be safe since there are no active members in an empty generation, so we just warn. + String message = t != null ? t.getMessage() : error.message(); + log.warn("Failed to write empty metadata for group {}: {}", group.groupId(), message); + } + }); + + List<Record> records = Collections.singletonList( + RecordHelpers.newGroupMetadataRecord(group, metadataImage.features().metadataVersion()) + ); + + return new CoordinatorResult<>(records, appendFuture); + + } else { + log.info("Stabilized group {} generation {} ({}-{}) with {} members", + groupId, group.generationId(), topicPartition.topic(), + topicPartition.partition(), group.size()); + + // Complete the awaiting join group response future for all the members after rebalancing + group.allMembers().forEach(member -> { + + List<JoinGroupResponseData.JoinGroupResponseMember> members = Collections.emptyList(); + if (group.isLeader(member.memberId())) { + members = group.currentGenericGroupMembers(); + } + + JoinGroupResponseData response = new JoinGroupResponseData() + .setMembers(members) + .setMemberId(member.memberId()) + .setGenerationId(group.generationId()) + .setProtocolName(group.protocolName().orElse(null)) + .setProtocolType(group.protocolType().orElse(null)) + .setLeader(group.leaderOrNull()) + .setSkipAssignment(false) + .setErrorCode(Errors.NONE.code()); + + group.completeJoinFuture(member, response); + rescheduleGenericGroupMemberHeartbeat(group, member); Review Comment: so that i understand: during completing rebalance, we schedule both pending sync (rebalance timeout) and heartbeats (session timeout). in practice, session timeout << rebalance timeout so heartbeats would expire and remove members. With cooperative rebalancing members should still be able to fetch records during completing rebalance phase. We want to extend the heartbeat here to rebalance timeout so that members are not removed by session timeout. is this correct? If so, i agree removing the heartbeat schedule sounds like the best approach since members are removed when pending sync expires anyways. -- 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