vamossagar12 commented on code in PR #14432:
URL: https://github.com/apache/kafka/pull/14432#discussion_r1405732141
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -849,21 +922,53 @@ 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;
+ 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);
+ if (memberEpoch == 0) {
+ // A new static member joins or the existing static member
rejoins.
+ if (member == null) {
+ // New static member.
+ member = group.getOrMaybeCreateMember(memberId,
createIfNotExists);
+ updatedMemberBuilder = new
ConsumerGroupMember.Builder(member);
+ log.info("[GroupId {}] Static member {} with instance id
{} joins the consumer group", groupId, memberId, instanceId);
+ } else {
+ // Static member rejoins with a different member id so it
should replace
+ // the previous instance iff the previous member had sent
a Leave group.
+ throwIfInstanceIdIsUnreleased(groupId, memberId,
instanceId, member);
+ // Replace the current member.
+ staticMemberReplaced = true;
+ updatedMemberBuilder = new
ConsumerGroupMember.Builder(memberId)
+ .setAssignedPartitions(member.assignedPartitions());
+ removeMemberAndCancelTimers(records, group.groupId(),
member.memberId());
Review Comment:
Added a log line
--
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
