guozhangwang commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r750719122
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##
@@ -755,18 +755,17 @@ protected void onJoinPrepare(int generation, String
memberId) {
@Override
public void onLeavePrepare() {
-// Save the current Generation and use that to get the memberId, as
the hb thread can change it at any time
+// Save the current Generation, as the hb thread can change it at any
time
final Generation currentGeneration = generation();
-final String memberId = currentGeneration.memberId;
-log.debug("Executing onLeavePrepare with generation {} and memberId
{}", currentGeneration, memberId);
Review comment:
What's the rationale of removing the member id in logging?
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##
@@ -755,18 +755,17 @@ protected void onJoinPrepare(int generation, String
memberId) {
@Override
public void onLeavePrepare() {
-// Save the current Generation and use that to get the memberId, as
the hb thread can change it at any time
+// Save the current Generation, as the hb thread can change it at any
time
final Generation currentGeneration = generation();
-final String memberId = currentGeneration.memberId;
-log.debug("Executing onLeavePrepare with generation {} and memberId
{}", currentGeneration, memberId);
+log.debug("Executing onLeavePrepare with generation {}",
currentGeneration);
// we should reset assignment and trigger the callback before leaving
group
Set droppedPartitions = new
HashSet<>(subscriptions.assignedPartitions());
if (subscriptions.hasAutoAssignedPartitions() &&
!droppedPartitions.isEmpty()) {
final Exception e;
-if (generation() == Generation.NO_GENERATION ||
rebalanceInProgress()) {
+if (currentGeneration.equals(Generation.NO_GENERATION) ||
rebalanceInProgress()) {
Review comment:
Good catch.
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##
@@ -612,7 +614,7 @@ public void handle(JoinGroupResponse joinResponse,
RequestFuture fut
// only need to reset the member id if generation has not been
changed,
// then retry immediately
if (generationUnchanged())
-resetGenerationOnResponseError(ApiKeys.JOIN_GROUP, error);
+resetGenerationOnResponseError(ApiKeys.JOIN_GROUP, error,
true);
Review comment:
nit: maybe it's now better to rename this function, to
`resetStateOnResponseError`?
--
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
