squah-confluent commented on code in PR #22708:
URL: https://github.com/apache/kafka/pull/22708#discussion_r3500693074
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -503,17 +503,23 @@ public void updateMember(StreamsGroupMember newMember) {
}
StreamsGroupMember oldMember = members.put(newMember.memberId(),
newMember);
maybeUpdateTaskProcessId(oldMember, newMember);
- updateStaticMember(newMember);
+ updateStaticMember(oldMember, newMember);
maybeUpdateGroupState();
endpointToPartitionsCache.remove(newMember.memberId());
}
/**
* Updates the member ID stored against the instance ID if the member is a
static member.
*
+ * @param oldMember The old member state.
* @param newMember The new member state.
*/
- private void updateStaticMember(StreamsGroupMember newMember) {
+ private void updateStaticMember(StreamsGroupMember oldMember,
StreamsGroupMember newMember) {
+ if (oldMember != null && oldMember.instanceId() != null &&
+ oldMember.instanceId().isPresent() &&
+ !oldMember.instanceId().equals(newMember.instanceId())) {
Review Comment:
Yes, `instanceId` is initialized to `null` on the replay path until the
first `StreamsGroupMemberMetadata` record.
--
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]
