jolshan commented on code in PR #13963: URL: https://github.com/apache/kafka/pull/13963#discussion_r1258653960
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java: ########## @@ -720,19 +765,116 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr ); if (!subscriptionMetadata.equals(group.subscriptionMetadata())) { - log.info("[GroupId " + groupId + "] Computed new subscription metadata: " + log.info("[GroupId " + group.groupId() + "] Computed new subscription metadata: " + subscriptionMetadata + "."); - records.add(newGroupSubscriptionMetadataRecord(groupId, subscriptionMetadata)); + records.add(newGroupSubscriptionMetadataRecord(group.groupId(), subscriptionMetadata)); } // We bump the group epoch. int groupEpoch = group.groupEpoch() + 1; - records.add(newGroupEpochRecord(groupId, groupEpoch)); + records.add(newGroupEpochRecord(group.groupId(), groupEpoch)); - return new CoordinatorResult<>(records, new ConsumerGroupHeartbeatResponseData() - .setMemberId(memberId) - .setMemberEpoch(-1) - ); + return records; + } + + /** + * Schedules (or reschedules) the session timeout for the member. + * + * @param groupId The group id. + * @param memberId The member id. + */ + private void scheduleConsumerGroupSessionTimeout( + String groupId, + String memberId + ) { + String key = consumerGroupSessionTimeoutKey(groupId, memberId); + timer.schedule(key, consumerGroupSessionTimeoutMs, TimeUnit.MILLISECONDS, true, () -> { + try { + ConsumerGroup group = getOrMaybeCreateConsumerGroup(groupId, false); + ConsumerGroupMember member = group.getOrMaybeCreateMember(memberId, false); + + log.info("[GroupId " + groupId + "] Member " + memberId + " fenced from the group because " + + "its session expired."); + + return consumerGroupFenceMember(group, member); + } catch (GroupIdNotFoundException ex) { + log.debug("[GroupId " + groupId + "] Could not fence " + memberId + " because the group " + + "does not exist."); + } catch (UnknownMemberIdException ex) { + log.debug("[GroupId " + groupId + "] Could not fence " + memberId + " because the member " + + "does not exist."); + } + + return Collections.emptyList(); + }); + } + + /** + * Cancels the session timeout of the member. + * + * @param groupId The group id. + * @param memberId The member id. + */ + private void cancelConsumerGroupSessionTimeout( + String groupId, + String memberId + ) { + timer.cancel(consumerGroupSessionTimeoutKey(groupId, memberId)); + } + + /** + * Schedules a revocation timeout for the member. + * + * @param groupId The group id. + * @param memberId The member id. + * @param revocationTimeoutMs The revocation timeout. + * @param expectedMemberEpoch The expected member epoch. + */ + private void scheduleConsumerGroupRevocationTimeout( + String groupId, + String memberId, + long revocationTimeoutMs, + int expectedMemberEpoch + ) { + String key = consumerGroupRevocationTimeoutKey(groupId, memberId); + timer.schedule(key, revocationTimeoutMs, TimeUnit.MILLISECONDS, true, () -> { + try { + ConsumerGroup group = getOrMaybeCreateConsumerGroup(groupId, false); + ConsumerGroupMember member = group.getOrMaybeCreateMember(memberId, false); + + if (member.state() != ConsumerGroupMember.MemberState.REVOKING && + member.memberEpoch() != expectedMemberEpoch) { Review Comment: We continue to revoke even if the epoch is unexpected? Also do we want to return early here? -- 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