jolshan commented on code in PR #15364:
URL: https://github.com/apache/kafka/pull/15364#discussion_r1520152684
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1400,35 +1440,35 @@ private void cancelConsumerGroupSessionTimeout(
}
/**
- * Schedules a revocation timeout for the member.
+ * Schedules a rebalance 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.
+ * @param memberEpoch The member epoch.
+ * @param rebalanceTimeoutMs The rebalance timeout.
*/
- private void scheduleConsumerGroupRevocationTimeout(
+ private void scheduleConsumerGroupRebalanceTimeout(
String groupId,
String memberId,
- long revocationTimeoutMs,
- int expectedMemberEpoch
+ int memberEpoch,
+ int rebalanceTimeoutMs
) {
- String key = consumerGroupRevocationTimeoutKey(groupId, memberId);
- timer.schedule(key, revocationTimeoutMs, TimeUnit.MILLISECONDS, true,
() -> {
+ String key = consumerGroupRebalanceTimeoutKey(groupId, memberId);
+ timer.schedule(key, rebalanceTimeoutMs, TimeUnit.MILLISECONDS, true,
() -> {
try {
ConsumerGroup group = getOrMaybeCreateConsumerGroup(groupId,
false);
ConsumerGroupMember member =
group.getOrMaybeCreateMember(memberId, false);
- if (member.state() != ConsumerGroupMember.MemberState.REVOKING
||
- member.memberEpoch() != expectedMemberEpoch) {
- log.debug("[GroupId {}] Ignoring revocation timeout for {}
because the member " +
- "state does not match the expected state.", groupId,
memberId);
+ if (member.memberEpoch() == memberEpoch) {
+ log.info("[GroupId {}] Member {} fenced from the group
because " +
+ "it failed to transition from epoch {} within
{}ms.",
+ groupId, memberId, memberEpoch, rebalanceTimeoutMs);
+ return new
CoordinatorResult<>(consumerGroupFenceMember(group, member));
Review Comment:
So it seems like this is one of the main changes here.
When we fence a group member, do we basically kick it out of the group and
force it to rejoin?
Can the client do this without restarting?
--
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
