vamossagar12 commented on code in PR #14432:
URL: https://github.com/apache/kafka/pull/14432#discussion_r1402217667
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -829,21 +902,50 @@ private
CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr
// Get or create the member.
if (memberId.isEmpty()) memberId = Uuid.randomUuid().toString();
- final ConsumerGroupMember member =
group.getOrMaybeCreateMember(memberId, createIfNotExists);
- throwIfMemberEpochIsInvalid(member, memberEpoch, ownedTopicPartitions);
-
- if (memberEpoch == 0) {
- log.info("[GroupId {}] Member {} joins the consumer group.",
groupId, memberId);
+ ConsumerGroupMember member;
+ ConsumerGroupMember.Builder updatedMemberBuilder;
+ ConsumerGroupMember updatedMember;
+ boolean staticMemberReplaced = false;
+ if (instanceId == null) {
+ member = group.getOrMaybeCreateMember(memberId, createIfNotExists);
+ throwIfMemberEpochIsInvalid(member, memberEpoch,
ownedTopicPartitions);
+ if (createIfNotExists) {
+ log.info("[GroupId {}] Member {} joins the consumer group.",
groupId, memberId);
+ }
+ updatedMemberBuilder = new ConsumerGroupMember.Builder(member);
+ } else {
+ member = group.staticMember(instanceId);
+ throwIfStaticMemberValidationFails(groupId, instanceId, member,
memberEpoch, memberId);
+ if (member == null) {
+ member = group.getOrMaybeCreateMember(memberId,
createIfNotExists);
+ }
+ throwIfMemberEpochIsInvalid(member, memberEpoch,
ownedTopicPartitions);
+ if (createIfNotExists) {
+ log.info("[GroupId {}] Member {}, Instance-Id {} joins the
consumer group.", groupId, memberId, instanceId);
+ }
+ staticMemberReplaced = staticMemberReplaced(memberEpoch, member);
+ if (staticMemberReplaced) {
+ removeMemberAndCancelTimers(records, group.groupId(),
member.memberId());
+ // The replacing member gets the same set of assignments as
the departed member.
+ updatedMemberBuilder = new
ConsumerGroupMember.Builder(memberId)
+ .setAssignedPartitions(member.assignedPartitions())
+
.setPartitionsPendingAssignment(member.partitionsPendingAssignment())
+
.setPartitionsPendingRevocation(member.partitionsPendingRevocation());
Review Comment:
If I don't set `setAssignedPartitions`, then the replacing static member
doesn't get it's assignments back. The pending assignments bit, I added just to
ensure all assignments from previous member are assigned to the new member. The
reason that happens is that it lands up
[here](https://github.com/apache/kafka/blob/6b26c0428a5cf1f64e458e5a33df3bbfb0e33c6b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/CurrentAssignmentBuilder.java#L175)
and if there are no partitions assigned, it gets back empty assignments.
--
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
