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 -- we don't validate
on the revoking state -- I guess we could be in a revoking state for the next
assignment...
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.
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?
--
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
