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?
--
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
