cadonna commented on code in PR #14879:
URL: https://github.com/apache/kafka/pull/14879#discussion_r1414403764
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -232,6 +235,16 @@ private void onResponse(final
ConsumerGroupHeartbeatResponse response, long curr
this.heartbeatRequestState.onSuccessfulAttempt(currentTimeMs);
this.heartbeatRequestState.resetTimer();
this.membershipManager.onHeartbeatResponseReceived(response.data());
+ if (previousGroupMetadataUdateEvent == null ||
+ previousGroupMetadataUdateEvent.memberEpoch() !=
membershipManager.memberEpoch()) {
+
+ final GroupMetadataUpdateEvent currentGroupMetadataUpdateEvent
= new GroupMetadataUpdateEvent(
+ membershipManager.memberEpoch(),
+ previousGroupMetadataUdateEvent == null ?
membershipManager.memberId() : previousGroupMetadataUdateEvent.memberId()
+ );
+
this.backgroundEventHandler.add(currentGroupMetadataUpdateEvent);
+ previousGroupMetadataUdateEvent =
currentGroupMetadataUpdateEvent;
+ }
Review Comment:
@dajac @lucasbru @kirktrue @AndrewJSchofield Could you please sanity check
this code snippet?
The heartbeat request manager should only send an group metadata update
event if it is the first heartbeat response it gets (i.e.
`previousGroupMetadataUdateEvent == null`), because that sets the member ID and
the member ID never changes after this. Or the member epoch changed in the last
heartbeat response (i.e., previousGroupMetadataUdateEvent.memberEpoch() !=
membershipManager.memberEpoch()).
Does this make sense?
--
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
