dajac commented on code in PR #15364:
URL: https://github.com/apache/kafka/pull/15364#discussion_r1521487635
##########
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:
> I may have asked this on a previous pr, but are we assuming the member
epoch of the member (not the one passed in) is always never less than the
member epoch passed into this method. That makes sense given the epoch is
monotonically increasing, but just wanted to confirm.
It is only newer when the member confirms that it has revoked the
partitions. Here we are basically saying that we want a member to transition
from its current epoch (and acks the revocation) within the rebalance timeout.
> As an aside, when we fence a group member, do we basically kick it out of
the group and force it to rejoin?
Can the client rejoin without restarting?
On the server, we only fence the member. On the client, it will rejoin when
it is fenced but this is a client side implementation choice.
--
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
