dajac commented on code in PR #18224: URL: https://github.com/apache/kafka/pull/18224#discussion_r1914517937
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -984,27 +984,32 @@ private boolean validateOnlineDowngradeWithReplacedMemberId( * Creates a ClassicGroup corresponding to the given ConsumerGroup. * * @param consumerGroup The converted ConsumerGroup. - * @param leavingMemberId The leaving member that triggers the downgrade validation. + * @param leavingMembers The leaving member(s) that triggered the downgrade validation. * @param joiningMember The newly joined member if the downgrade is triggered by static member replacement. + * When not null, must have an instanceId that matches an existing member. * @param records The record list to which the conversion records are added. */ private void convertToClassicGroup( ConsumerGroup consumerGroup, - String leavingMemberId, + Set<ConsumerGroupMember> leavingMembers, ConsumerGroupMember joiningMember, List<CoordinatorRecord> records ) { if (joiningMember == null) { consumerGroup.createGroupTombstoneRecords(records); } else { - consumerGroup.createGroupTombstoneRecordsWithReplacedMember(records, leavingMemberId, joiningMember.memberId()); + // We've already generated the records to replace replacedMember with joiningMember, + // so we need to tombstone joiningMember instead. + ConsumerGroupMember replacedMember = consumerGroup.staticMember(joiningMember.instanceId()); + throwIfStaticMemberIsUnknown(replacedMember, joiningMember.instanceId()); Review Comment: I wonder if we should rather throw an illegal argument exception if joiningMember is not a static member. What do you think? ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java: ########## @@ -1255,12 +1255,27 @@ public boolean supportsClassicProtocols(String memberProtocolType, Set<String> m /** * Checks whether all the members use the classic protocol except the given member. * - * @param memberId The member to remove. + * @param member The member to remove. * @return A boolean indicating whether all the members use the classic protocol. */ - public boolean allMembersUseClassicProtocolExcept(String memberId) { - return numClassicProtocolMembers() == members().size() - 1 && - !getOrMaybeCreateMember(memberId, false).useClassicProtocol(); + public boolean allMembersUseClassicProtocolExcept(ConsumerGroupMember member) { + return numClassicProtocolMembers() == members().size() - 1 && !member.useClassicProtocol(); + } + + /** + * Checks whether all the members use the classic protocol except the given members. + * + * @param members The members to remove. + * @return A boolean indicating whether all the members use the classic protocol. + */ + public boolean allMembersUseClassicProtocolExcept(Set<ConsumerGroupMember> members) { Review Comment: Should we add a unit test for this new method? ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java: ########## @@ -1412,7 +1412,11 @@ public static ClassicGroup fromConsumerGroup( // If the downgraded is triggered by the joining static member replacing // the leaving static member, the joining member should take the assignment // of the leaving one. - memberId = leavingMemberId; + ConsumerGroupMember replacedMember = consumerGroup.staticMember(joiningMember.instanceId()); + if (replacedMember == null) { + throw Errors.UNKNOWN_MEMBER_ID.exception("Instance id " + joiningMember.instanceId() + " is unknown."); + } Review Comment: ditto about the exception. -- 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