squah-confluent commented on code in PR #18224:
URL: https://github.com/apache/kafka/pull/18224#discussion_r1916133852
##########
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 don't feel strongly about this. The exception is closer to a programmer
error than due to a bad client request, so an illegal argument exception could
make more sense. I've updated the code.
--
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
