jolshan commented on code in PR #13963:
URL: https://github.com/apache/kafka/pull/13963#discussion_r1258644974
##########
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:
based on the previous pr, we retry these exceptions. I can imagine that some
metadata was slow to update or something and eventually it could succeed. Do we
have any path forward though if this request was issued and the group/member is
no longer there? I guess this relying on canceling the task when we call group
leave for the member?
--
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
