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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
