guozhangwang commented on a change in pull request #8834: URL: https://github.com/apache/kafka/pull/8834#discussion_r480425820
########## File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java ########## @@ -497,40 +501,18 @@ private synchronized void resetStateAndRejoin() { joinFuture.addListener(new RequestFutureListener<ByteBuffer>() { @Override public void onSuccess(ByteBuffer value) { - // handle join completion in the callback so that the callback will be invoked Review comment: Well I should say part of that (the enabling of the heartbeat thread) is in JoinGroup response handler, while the rest (update metrics, etc) is in SyncGroup response handler. ########## File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala ########## @@ -639,7 +641,11 @@ class GroupCoordinator(val brokerId: Int, responseCallback(Errors.UNKNOWN_MEMBER_ID) case CompletingRebalance => - responseCallback(Errors.REBALANCE_IN_PROGRESS) + // consumers may start sending heartbeat after join-group response, in which case + // we should treat them as normal hb request and reset the timer + val member = group.get(memberId) Review comment: It would return the error code before: that is because it does not expect clients to send heartbeat before sending sync-group requests. Now it is not the case any more. ########## File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala ########## @@ -639,7 +641,11 @@ class GroupCoordinator(val brokerId: Int, responseCallback(Errors.UNKNOWN_MEMBER_ID) case CompletingRebalance => - responseCallback(Errors.REBALANCE_IN_PROGRESS) Review comment: I had a discussion with @hachikuji about this. I think logically it should not return `REBALANCE_IN_PROGRESS` and clients in the future should update its handling logic too, maybe after some releases where we can break client-broker compatibility. ########## 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(); resetJoinGroupFuture(); - return false; } } else { final RuntimeException exception = future.exception(); - log.info("Join group failed with {}", exception.toString()); + log.info("Rebalance failed with {}", exception.toString()); Review comment: The reason I changed it is exactly that it may not always due to join-group :) If sync-group failed, this could also be triggered. ---------------------------------------------------------------- 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