dajac commented on code in PR #15364: URL: https://github.com/apache/kafka/pull/15364#discussion_r1502633032
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java: ########## @@ -2079,253 +2056,20 @@ public void testReconciliationProcess() { assertRecordsEquals(Collections.singletonList( RecordHelpers.newCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId3) + .setState(MemberState.STABLE) .setMemberEpoch(11) .setPreviousMemberEpoch(11) - .setTargetMemberEpoch(11) .setAssignedPartitions(mkAssignment( mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 1))) .build())), result.records() ); - assertEquals(ConsumerGroupMember.MemberState.STABLE, context.consumerGroupMemberState(groupId, memberId3)); + assertEquals(MemberState.STABLE, context.consumerGroupMemberState(groupId, memberId3)); assertEquals(ConsumerGroup.ConsumerGroupState.STABLE, context.consumerGroupState(groupId)); } - @Test - public void testReconciliationRestartsWhenNewTargetAssignmentIsInstalled() { Review Comment: Correct. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -1211,13 +1192,99 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr // 1. The member reported its owned partitions; // 2. The member just joined or rejoined to group (epoch equals to zero); // 3. The member's assignment has been updated. - if (ownedTopicPartitions != null || memberEpoch == 0 || assignmentUpdated) { + if (ownedTopicPartitions != null || memberEpoch == 0 || hasAssignedPartitionsChanged(member, updatedMember)) { response.setAssignment(createResponseAssignment(updatedMember)); } return new CoordinatorResult<>(records, response); } + /** + * Reconciles the current assignment of the member if needed. + * + * @param groupId The group id. + * @param member The member to reconcile. + * @param currentPartitionEpoch The function returning the current epoch of + * a given partition. + * @param targetAssignmentEpoch The target assignment epoch. + * @param targetAssignment The target assignment. + * @param ownedTopicPartitions The list of partitions owned by the member. This + * is reported in the ConsumerGroupHeartbeat API and + * it could be null if not provided. + * @param records The list to accumulate any new records. + * @return The received member if no changes have been made; or a new + * member containing the new assignment. + */ + private ConsumerGroupMember maybeReconcile( + String groupId, + ConsumerGroupMember member, + BiFunction<Uuid, Integer, Integer> currentPartitionEpoch, + int targetAssignmentEpoch, + Assignment targetAssignment, + List<ConsumerGroupHeartbeatRequestData.TopicPartitions> ownedTopicPartitions, + List<Record> records + ) { + if (member.isReconciledTo(targetAssignmentEpoch)) { + return member; + } + + ConsumerGroupMember updatedMember = new CurrentAssignmentBuilder(member) + .withTargetAssignment(targetAssignmentEpoch, targetAssignment) + .withCurrentPartitionEpoch(currentPartitionEpoch) + .withOwnedTopicPartitions(ownedTopicPartitions) + .build(); + + if (!updatedMember.equals(member)) { + records.add(newCurrentAssignmentRecord(groupId, updatedMember)); + + log.info("[GroupId {}] Member {} new assignment state: epoch={}, previousEpoch={}, state={}, " + + "assignedPartitions={} and revokedPartitions={}.", + groupId, updatedMember.memberId(), updatedMember.memberEpoch(), updatedMember.previousMemberEpoch(), updatedMember.state(), + formatAssignment(updatedMember.assignedPartitions()), formatAssignment(updatedMember.revokedPartitions())); + + if (updatedMember.state() == MemberState.UNREVOKED_PARTITIONS) { + scheduleConsumerGroupRebalanceTimeout( + groupId, + updatedMember.memberId(), + updatedMember.memberEpoch(), + updatedMember.rebalanceTimeoutMs() + ); + } else { Review Comment: Yeah, that's right. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/CurrentAssignmentBuilder.java: ########## @@ -170,72 +127,119 @@ public CurrentAssignmentBuilder withOwnedTopicPartitions( * @return A new ConsumerGroupMember or the current one. */ public ConsumerGroupMember build() { - // A new target assignment has been installed, we need to restart - // the reconciliation loop from the beginning. - if (targetAssignmentEpoch != member.targetMemberEpoch()) { - return transitionToNewTargetAssignmentState(); - } - switch (member.state()) { - // Check if the partitions have been revoked by the member. - case REVOKING: - return maybeTransitionFromRevokingToAssigningOrStable(); + case STABLE: + // When the member is in the STABLE state, we verify if a newer + // epoch (or target assignment) is available. If it is, we can + // reconcile the member towards it. Otherwise, we return. + if (member.memberEpoch() != targetAssignmentEpoch) { + return computeNextAssignment( + member.memberEpoch(), + member.assignedPartitions() + ); + } else { + return member; + } - // Check if pending partitions have been freed up. - case ASSIGNING: - return maybeTransitionFromAssigningToAssigningOrStable(); + case UNREVOKED_PARTITIONS: + // When the member is in the UNREVOKED_PARTITIONS state, we wait + // until the member has revoked the necessary partitions. They are + // considered revoked when they are not anymore reported in the + // owned partitions set in the ConsumerGroupHeartbeat API. - // Nothing to do. - case STABLE: - return member; + // If the member does not provide its owned partitions. We cannot + // progress. + if (ownedTopicPartitions == null) { + return member; + } + + // If the member provides its owned partitions. We verify if it still + // owens any of the revoked partitions. If it does, we cannot progress. + for (ConsumerGroupHeartbeatRequestData.TopicPartitions topicPartitions : ownedTopicPartitions) { + for (Integer partitionId : topicPartitions.partitions()) { + boolean stillHasRevokedPartition = member + .revokedPartitions() + .getOrDefault(topicPartitions.topicId(), Collections.emptySet()) + .contains(partitionId); + if (stillHasRevokedPartition) { + return member; + } + } + } + + // When the member has revoked all the pending partitions, it can + // transition to the next epoch (current + 1) and we can reconcile + // its state towards the latest target assignment. + return computeNextAssignment( + member.memberEpoch() + 1, Review Comment: Yep. You got it. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -1168,36 +1170,15 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr // 3. Reconcile the member's assignment with the target assignment. This is only required if // the member is not stable or if a new target assignment has been installed. Review Comment: I changed the entire comment. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java: ########## @@ -1807,9 +1776,9 @@ public void testReconciliationProcess() { assertRecordsEquals(Collections.singletonList( RecordHelpers.newCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId1) + .setState(MemberState.UNREVOKED_PARTITIONS) .setMemberEpoch(10) - .setPreviousMemberEpoch(9) - .setTargetMemberEpoch(11) + .setPreviousMemberEpoch(10) Review Comment: In the [previous implementation](https://github.com/apache/kafka/pull/15364/files#diff-6c5e23803064f4b7a122eee29736d036b9cfe244e69f48751ba163d62e2bf35fL257), the previous epoch was not updated. It feels like a bug. In the new implementation, I have set the previous epoch whenever the member epoch is updated. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -1211,13 +1192,99 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr // 1. The member reported its owned partitions; Review Comment: No. This condition is actually slightly incorrect. I will fix it. The idea is to provide a full response when we received a full request from the client. Here, full means that all the fields are set. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -1211,13 +1192,99 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr // 1. The member reported its owned partitions; // 2. The member just joined or rejoined to group (epoch equals to zero); // 3. The member's assignment has been updated. - if (ownedTopicPartitions != null || memberEpoch == 0 || assignmentUpdated) { + if (ownedTopicPartitions != null || memberEpoch == 0 || hasAssignedPartitionsChanged(member, updatedMember)) { response.setAssignment(createResponseAssignment(updatedMember)); } return new CoordinatorResult<>(records, response); } + /** + * Reconciles the current assignment of the member if needed. + * + * @param groupId The group id. + * @param member The member to reconcile. + * @param currentPartitionEpoch The function returning the current epoch of + * a given partition. + * @param targetAssignmentEpoch The target assignment epoch. + * @param targetAssignment The target assignment. + * @param ownedTopicPartitions The list of partitions owned by the member. This + * is reported in the ConsumerGroupHeartbeat API and + * it could be null if not provided. + * @param records The list to accumulate any new records. + * @return The received member if no changes have been made; or a new + * member containing the new assignment. + */ + private ConsumerGroupMember maybeReconcile( + String groupId, + ConsumerGroupMember member, + BiFunction<Uuid, Integer, Integer> currentPartitionEpoch, + int targetAssignmentEpoch, + Assignment targetAssignment, + List<ConsumerGroupHeartbeatRequestData.TopicPartitions> ownedTopicPartitions, + List<Record> records + ) { + if (member.isReconciledTo(targetAssignmentEpoch)) { + return member; + } + + ConsumerGroupMember updatedMember = new CurrentAssignmentBuilder(member) + .withTargetAssignment(targetAssignmentEpoch, targetAssignment) + .withCurrentPartitionEpoch(currentPartitionEpoch) + .withOwnedTopicPartitions(ownedTopicPartitions) + .build(); + + if (!updatedMember.equals(member)) { + records.add(newCurrentAssignmentRecord(groupId, updatedMember)); + + log.info("[GroupId {}] Member {} new assignment state: epoch={}, previousEpoch={}, state={}, " + + "assignedPartitions={} and revokedPartitions={}.", + groupId, updatedMember.memberId(), updatedMember.memberEpoch(), updatedMember.previousMemberEpoch(), updatedMember.state(), + formatAssignment(updatedMember.assignedPartitions()), formatAssignment(updatedMember.revokedPartitions())); + + if (updatedMember.state() == MemberState.UNREVOKED_PARTITIONS) { + scheduleConsumerGroupRebalanceTimeout( + groupId, + updatedMember.memberId(), + updatedMember.memberEpoch(), + updatedMember.rebalanceTimeoutMs() + ); + } else { + cancelConsumerGroupRebalanceTimeout(groupId, updatedMember.memberId()); + } + } + + return updatedMember; + } + + private String formatAssignment( Review Comment: We could get it from the MetadataImage but it may not be available. I think that using the topic id is fine here. ########## group-coordinator/src/main/resources/common/message/ConsumerGroupCurrentMemberAssignmentValue.json: ########## @@ -24,20 +24,12 @@ "about": "The current member epoch that is expected from the member in the heartbeat request." }, { "name": "PreviousMemberEpoch", "versions": "0+", "type": "int32", "about": "If the last epoch bump is lost before reaching the member, the member will retry with the previous epoch." }, - { "name": "TargetMemberEpoch", "versions": "0+", "type": "int32", - "about": "The target epoch corresponding to the assignment used to compute the AssignedPartitions, the PartitionsPendingRevocation and the PartitionsPendingAssignment fields." }, + { "name": "State", "versions": "0+", "type": "int8", + "about": "The member state. See ConsumerGroupMember.MemberState for the possible values." }, { "name": "AssignedPartitions", "versions": "0+", "type": "[]TopicPartitions", "about": "The partitions assigned to (or owned by) this member." }, { "name": "PartitionsPendingRevocation", "versions": "0+", "type": "[]TopicPartitions", - "about": "The partitions that must be revoked by this member." }, - { "name": "PartitionsPendingAssignment", "versions": "0+", "type": "[]TopicPartitions", - "about": "The partitions that will be assigned to this member when they are freed up by their current owners." }, - { "name": "Error", "versions": "0+", "type": "int8", - "about": "The error reported by the assignor." }, - { "name": "MetadataVersion", "versions": "0+", "type": "int16", - "about": "The version of the metadata bytes." }, - { "name": "MetadataBytes", "versions": "0+", "type": "bytes", - "about": "The metadata bytes." } Review Comment: I thought that we could remove unused fields as we are making non-backward compatible changes anyway. Those fields will be used when we implement the client side assignor, if we ever do it. -- 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