jolshan commented on code in PR #13963:
URL: https://github.com/apache/kafka/pull/13963#discussion_r1258681834
##########
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) {
Review Comment:
Oh I see. We only catch the exceptions in the other PR if they are thrown
(not caught) 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
