dajac commented on code in PR #18224:
URL: https://github.com/apache/kafka/pull/18224#discussion_r1910341356
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java:
##########
@@ -1368,7 +1367,8 @@ public static ClassicGroup fromConsumerGroup(
);
consumerGroup.members().forEach((memberId, member) -> {
- if (!memberId.equals(leavingMemberId)) {
+ if (!leavingMembers.contains(member) &&
+ (joiningMember == null ||
!joiningMember.instanceId().equals(member.instanceId()))) {
Review Comment:
`joiningMember.instanceId()` maybe null here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -984,27 +984,31 @@ 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 replacedMemberId
with joiningMember,
+ // so we need to tombstone joiningMember instead.
+ String replacedMemberId =
consumerGroup.staticMember(joiningMember.instanceId()).memberId();
Review Comment:
ditto.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java:
##########
@@ -1412,7 +1412,7 @@ 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;
+ memberId =
consumerGroup.staticMember(joiningMember.instanceId()).memberId();
Review Comment:
ditto.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
