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

Reply via email to