guozhangwang commented on a change in pull request #8834:
URL: https://github.com/apache/kafka/pull/8834#discussion_r485324085
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -446,14 +453,15 @@ boolean joinGroupIfNeeded(final Timer timer) {
resetJoinGroupFuture();
needsJoinPrepare = true;
} else {
- log.info("Generation data was cleared by heartbeat thread.
Initiating rejoin.");
+ log.info("Generation data was cleared by heartbeat thread
to {} and state is now {} before " +
+ "the rebalance callback is triggered, marking this
rebalance as failed and retry",
+ generation, state);
resetStateAndRejoin();
Review comment:
That's a good question. I just thought about this and I think I can
change the caller of `resetGeneration` (which is the only place that hb thread
can reset the generation) and move the `state = MemberState.UNJOINED;` into the
callee to make sure that they are always changed together.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -433,7 +440,7 @@ boolean joinGroupIfNeeded(final Timer timer) {
generationSnapshot = this.generation;
}
- if (generationSnapshot != Generation.NO_GENERATION) {
+ if (generationSnapshot != Generation.NO_GENERATION && state ==
MemberState.STABLE) {
Review comment:
Ack, will do.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
