From 5e404e2fcdd947a4087d1a5991f95957b3bddfdc Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Mon, 6 Apr 2020 16:46:53 -0700 Subject: [PATCH 1/9] first pass --- .../internals/ConsumerCoordinator.java | 7 ++ .../coordinator/group/GroupCoordinator.scala | 100 ++++++++++-------- .../coordinator/group/GroupMetadata.scala | 8 +- 3 files changed, 69 insertions(+), 46 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index b8d44b47e9520..d551148886a90 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -349,6 +349,13 @@ protected void onJoinComplete(int generation, Set ownedPartitions = new HashSet<>(subscriptions.assignedPartitions()); + // should at least encode the short version + if (assignmentBuffer.remaining() < 2) + throw new IllegalStateException("There is insufficient bytes available to read assignment from the sync-group response (" + + "actual byte size " + assignmentBuffer.remaining() + ") , this is not expected; " + + "it is possible that the leader's assign function is buggy and did not return any assignment for this member, " + + "or because static member is configured and the protocol is buggy hence did not get the assignment for this member"); + Assignment assignment = ConsumerProtocol.deserializeAssignment(assignmentBuffer); Set assignedPartitions = new HashSet<>(assignment.partitions()); diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index e4525de3a9d38..772d51d19f5c4 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -202,41 +202,7 @@ class GroupCoordinator(val brokerId: Int, val newMemberId = group.generateMemberId(clientId, groupInstanceId) if (group.hasStaticMember(groupInstanceId)) { - val oldMemberId = group.getStaticMemberId(groupInstanceId) - info(s"Static member $groupInstanceId of group ${group.groupId} with unknown member id rejoins, assigning new member id $newMemberId, while " + - s"old member id $oldMemberId will be removed.") - - val currentLeader = group.leaderOrNull - val member = group.replaceGroupInstance(oldMemberId, newMemberId, groupInstanceId) - // Heartbeat of old member id will expire without effect since the group no longer contains that member id. - // New heartbeat shall be scheduled with new member id. - completeAndScheduleNextHeartbeatExpiration(group, member) - - val knownStaticMember = group.get(newMemberId) - group.updateMember(knownStaticMember, protocols, responseCallback) - - group.currentState match { - case Stable | CompletingRebalance => - info(s"Static member joins during ${group.currentState} stage will not trigger rebalance.") - group.maybeInvokeJoinCallback(member, JoinGroupResult( - members = List.empty, - memberId = newMemberId, - generationId = group.generationId, - protocolType = group.protocolType, - protocolName = group.protocolName, - // We want to avoid current leader performing trivial assignment while the group - // is in stable/awaiting sync stage, because the new assignment in leader's next sync call - // won't be broadcast by a stable/awaiting sync group. This could be guaranteed by - // always returning the old leader id so that the current leader won't assume itself - // as a leader based on the returned message, since the new member.id won't match - // returned leader id, therefore no assignment will be performed. - leaderId = currentLeader, - error = Errors.NONE)) - case Empty | Dead => - throw new IllegalStateException(s"Group ${group.groupId} was not supposed to be " + - s"in the state ${group.currentState} when the unknown static member $groupInstanceId rejoins.") - case PreparingRebalance => - } + updateStaticMemberAndRebalance(group, newMemberId, groupInstanceId, protocols, responseCallback) } else if (requireKnownMemberId) { // If member id required (dynamic membership), register the member in the pending member list // and send back a response to call for another join group request with allocated member id. @@ -246,7 +212,7 @@ class GroupCoordinator(val brokerId: Int, addPendingMemberExpiration(group, newMemberId, sessionTimeoutMs) responseCallback(JoinGroupResult(newMemberId, Errors.MEMBER_ID_REQUIRED)) } else { - debug(s"${if (groupInstanceId.isDefined) "Static" else "Dynamic"} Member with unknown member id joins group ${group.groupId} in " + + info(s"${if (groupInstanceId.isDefined) "Static" else "Dynamic"} Member with unknown member id joins group ${group.groupId} in " + s"${group.currentState} state. Created a new member id $newMemberId for this member and add to the group.") addMemberAndRebalance(rebalanceTimeoutMs, sessionTimeoutMs, newMemberId, groupInstanceId, clientId, clientHost, protocolType, protocols, group, responseCallback) @@ -287,7 +253,7 @@ class GroupCoordinator(val brokerId: Int, } } else { val groupInstanceIdNotFound = groupInstanceId.isDefined && !group.hasStaticMember(groupInstanceId) - if (group.isStaticMemberFenced(memberId, groupInstanceId)) { + if (group.isStaticMemberFenced(memberId, groupInstanceId, "join-group")) { // given member id doesn't match with the groupInstanceId. Inform duplicate instance to shut down immediately. responseCallback(JoinGroupResult(memberId, Errors.FENCED_INSTANCE_ID)) } else if (!group.has(memberId) || groupInstanceIdNotFound) { @@ -397,7 +363,7 @@ class GroupCoordinator(val brokerId: Int, // coordinator OR the group is in a transient unstable phase. Let the member retry // finding the correct coordinator and rejoin. responseCallback(SyncGroupResult(Errors.COORDINATOR_NOT_AVAILABLE)) - } else if (group.isStaticMemberFenced(memberId, groupInstanceId)) { + } else if (group.isStaticMemberFenced(memberId, groupInstanceId, "sync-group")) { responseCallback(SyncGroupResult(Errors.FENCED_INSTANCE_ID)) } else if (!group.has(memberId)) { responseCallback(SyncGroupResult(Errors.UNKNOWN_MEMBER_ID)) @@ -483,7 +449,7 @@ class GroupCoordinator(val brokerId: Int, val memberId = leavingMember.memberId val groupInstanceId = Option(leavingMember.groupInstanceId) if (memberId != JoinGroupRequest.UNKNOWN_MEMBER_ID - && group.isStaticMemberFenced(memberId, groupInstanceId)) { + && group.isStaticMemberFenced(memberId, groupInstanceId, "leave-group")) { memberLeaveError(leavingMember, Errors.FENCED_INSTANCE_ID) } else if (group.isPendingMember(memberId)) { if (groupInstanceId.isDefined) { @@ -640,7 +606,7 @@ class GroupCoordinator(val brokerId: Int, // coordinator OR the group is in a transient unstable phase. Let the member retry // finding the correct coordinator and rejoin. responseCallback(Errors.COORDINATOR_NOT_AVAILABLE) - } else if (group.isStaticMemberFenced(memberId, groupInstanceId)) { + } else if (group.isStaticMemberFenced(memberId, groupInstanceId, "heartbeat")) { responseCallback(Errors.FENCED_INSTANCE_ID) } else if (!group.has(memberId)) { responseCallback(Errors.UNKNOWN_MEMBER_ID) @@ -739,7 +705,7 @@ class GroupCoordinator(val brokerId: Int, // coordinator OR the group is in a transient unstable phase. Let the member retry // finding the correct coordinator and rejoin. responseCallback(offsetMetadata.map { case (k, _) => k -> Errors.COORDINATOR_NOT_AVAILABLE }) - } else if (group.isStaticMemberFenced(memberId, groupInstanceId)) { + } else if (group.isStaticMemberFenced(memberId, groupInstanceId, "txn-commit-offsets")) { responseCallback(offsetMetadata.map { case (k, _) => k -> Errors.FENCED_INSTANCE_ID }) } else if (memberId != JoinGroupRequest.UNKNOWN_MEMBER_ID && !group.has(memberId)) { // Enforce member id when it is set. @@ -766,7 +732,7 @@ class GroupCoordinator(val brokerId: Int, // coordinator OR the group is in a transient unstable phase. Let the member retry // finding the correct coordinator and rejoin. responseCallback(offsetMetadata.map { case (k, _) => k -> Errors.COORDINATOR_NOT_AVAILABLE }) - } else if (group.isStaticMemberFenced(memberId, groupInstanceId)) { + } else if (group.isStaticMemberFenced(memberId, groupInstanceId, "commit-offsets")) { responseCallback(offsetMetadata.map { case (k, _) => k -> Errors.FENCED_INSTANCE_ID }) } else if (generationId < 0 && group.is(Empty)) { // The group is only using Kafka to store offsets. @@ -1025,7 +991,55 @@ class GroupCoordinator(val brokerId: Int, } else { group.removePendingMember(memberId) } - maybePrepareRebalance(group, s"Adding new member $memberId with group instanceid $groupInstanceId") + maybePrepareRebalance(group, s"Adding new member $memberId with group instance id $groupInstanceId") + } + + private def updateStaticMemberAndRebalance(group: GroupMetadata, + newMemberId: String, + groupInstanceId: Option[String], + protocols: List[(String, Array[Byte])], + responseCallback: JoinCallback): Unit = { + val oldMemberId = group.getStaticMemberId(groupInstanceId) + info(s"Static member $groupInstanceId of group ${group.groupId} with unknown member id rejoins, assigning new member id $newMemberId, while " + + s"old member id $oldMemberId will be removed.") + + val currentLeader = group.leaderOrNull + val member = group.replaceGroupInstance(oldMemberId, newMemberId, groupInstanceId) + // Heartbeat of old member id will expire without effect since the group no longer contains that member id. + // New heartbeat shall be scheduled with new member id. + completeAndScheduleNextHeartbeatExpiration(group, member) + + val knownStaticMember = group.get(newMemberId) + group.updateMember(knownStaticMember, protocols, responseCallback) + + group.currentState match { + case Stable => + info(s"Static member joins during Stable stage will not trigger rebalance.") + group.maybeInvokeJoinCallback(member, JoinGroupResult( + members = List.empty, + memberId = newMemberId, + generationId = group.generationId, + protocolType = group.protocolType, + protocolName = group.protocolName, + // We want to avoid current leader performing trivial assignment while the group + // is in stable stage, because the new assignment in leader's next sync call + // won't be broadcast by a stable group. This could be guaranteed by + // always returning the old leader id so that the current leader won't assume itself + // as a leader based on the returned message, since the new member.id won't match + // returned leader id, therefore no assignment will be performed. + leaderId = currentLeader, + error = Errors.NONE)) + case CompletingRebalance => + // if the group is in after-sync stage, upon getting a new join-group of a known static member + // we should still trigger a new rebalance, since the old member may already be sent to the leader + // for assignment, and hence when the assignment gets back there would be a mismatch of the old member id + // with the new replaced member id. As a result the new member id would not get any assignment. + prepareRebalance(group, s"Updating metadata for static member ${member.memberId} with instance id $groupInstanceId") + case Empty | Dead => + throw new IllegalStateException(s"Group ${group.groupId} was not supposed to be " + + s"in the state ${group.currentState} when the unknown static member $groupInstanceId rejoins.") + case PreparingRebalance => + } } private def updateMemberAndRebalance(group: GroupMetadata, diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala index 44b33d3029f99..2688cd4a86c35 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala @@ -385,11 +385,13 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState * 2. group stored member.id doesn't match with given member.id */ def isStaticMemberFenced(memberId: String, - groupInstanceId: Option[String]): Boolean = { + groupInstanceId: Option[String], + operation: String): Boolean = { if (hasStaticMember(groupInstanceId) && getStaticMemberId(groupInstanceId) != memberId) { - error(s"given member.id $memberId is identified as a known static member ${groupInstanceId.get}," + - s"but not matching the expected member.id ${getStaticMemberId(groupInstanceId)}") + error(s"given member.id $memberId is identified as a known static member ${groupInstanceId.get}, " + + s"but not matching the expected member.id ${getStaticMemberId(groupInstanceId)} during $operation, will " + + s"respond with instance fenced error") true } else false From 0df8dc91fa550a5bc22978678431528c7011df35 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 7 Apr 2020 11:57:17 -0700 Subject: [PATCH 2/9] copy-past all changes --- .../internals/AbstractCoordinator.java | 37 ++++++++++----- .../internals/ConsumerCoordinator.java | 20 +++++--- .../clients/consumer/internals/Heartbeat.java | 37 +++++++++------ .../internals/AbstractCoordinatorTest.java | 46 +++++++++++++++++++ .../kafka/api/PlaintextConsumerTest.scala | 6 ++- .../group/GroupCoordinatorTest.scala | 24 ++++++---- 6 files changed, 129 insertions(+), 41 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index af67ab57c641c..4844922c64afe 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -451,8 +451,9 @@ boolean joinGroupIfNeeded(final Timer timer) { return false; } } else { - resetJoinGroupFuture(); final RuntimeException exception = future.exception(); + log.info("Join group failed with {}", exception.toString()); + resetJoinGroupFuture(); if (exception instanceof UnknownMemberIdException || exception instanceof RebalanceInProgressException || exception instanceof IllegalGenerationException || @@ -889,17 +890,26 @@ protected synchronized String memberId() { } private synchronized void resetGeneration() { - this.generation = Generation.NO_GENERATION; - resetStateAndRejoin(); + rejoinNeeded = true; + generation = Generation.NO_GENERATION; } synchronized void resetGenerationOnResponseError(ApiKeys api, Errors error) { - log.debug("Resetting generation after encountering {} from {} response", error, api); + log.debug("Resetting generation after encountering {} from {} response and requesting re-join", error, api); + + // only reset the state to un-joined when it is not already in rebalancing + if (state != MemberState.REBALANCING) + state = MemberState.UNJOINED; + resetGeneration(); } synchronized void resetGenerationOnLeaveGroup() { log.debug("Resetting generation due to consumer pro-actively leaving the group"); + + // always set the state to un-joined + state = MemberState.UNJOINED; + resetGeneration(); } @@ -1014,7 +1024,8 @@ public void handle(LeaveGroupResponse leaveResponse, RequestFuture future) // visible for testing synchronized RequestFuture sendHeartbeatRequest() { - log.debug("Sending Heartbeat request to coordinator {}", coordinator); + log.debug("Sending Heartbeat request with generation {} and member id {} to coordinator {}", + generation.generationId, generation.memberId, coordinator); HeartbeatRequest.Builder requestBuilder = new HeartbeatRequest.Builder(new HeartbeatRequestData() .setGroupId(rebalanceConfig.groupId) @@ -1022,10 +1033,16 @@ synchronized RequestFuture sendHeartbeatRequest() { .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null)) .setGenerationId(this.generation.generationId)); return client.send(coordinator, requestBuilder) - .compose(new HeartbeatResponseHandler()); + .compose(new HeartbeatResponseHandler(generation)); } private class HeartbeatResponseHandler extends CoordinatorResponseHandler { + private final Generation sentGeneration; + + private HeartbeatResponseHandler(final Generation generation) { + this.sentGeneration = generation; + } + @Override public void handle(HeartbeatResponse heartbeatResponse, RequestFuture future) { sensors.heartbeatSensor.record(response.requestLatencyMs()); @@ -1035,7 +1052,7 @@ public void handle(HeartbeatResponse heartbeatResponse, RequestFuture futu future.complete(null); } else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) { - log.info("Attempt to heartbeat failed since coordinator {} is either not started or not valid.", + log.info("Attempt to heartbeat failed since coordinator {} is either not started or not valid", coordinator()); markCoordinatorUnknown(); future.raise(error); @@ -1044,14 +1061,14 @@ public void handle(HeartbeatResponse heartbeatResponse, RequestFuture futu requestRejoin(); future.raise(error); } else if (error == Errors.ILLEGAL_GENERATION) { - log.info("Attempt to heartbeat failed since generation {} is not current", generation.generationId); + log.info("Attempt to heartbeat failed since generation {} is not current", sentGeneration.generationId); resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); future.raise(error); } else if (error == Errors.FENCED_INSTANCE_ID) { log.error("Received fatal exception: group.instance.id gets fenced"); future.raise(error); } else if (error == Errors.UNKNOWN_MEMBER_ID) { - log.info("Attempt to heartbeat failed since member id {} is not valid.", generation.memberId); + log.info("Attempt to heartbeat failed since member id {} is not valid.", sentGeneration.memberId); resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); future.raise(error); } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) { @@ -1432,6 +1449,4 @@ final boolean hasUnknownGeneration() { final boolean hasValidMemberId() { return generation != Generation.NO_GENERATION && generation.hasMemberId(); } - - } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index b0f476a6157a5..b2bb2345e0459 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -272,6 +272,18 @@ private void maybeUpdateJoinedSubscription(Set assignedPartition } } + private Exception invokeOnAssignment(final ConsumerPartitionAssignor assignor, final Assignment assignment) { + log.info("Notifying assignor about the new {}", assignment); + + try { + assignor.onAssignment(assignment, groupMetadata); + } catch (Exception e) { + return e; + } + + return null; + } + private Exception invokePartitionsAssigned(final Set assignedPartitions) { log.info("Adding newly assigned partitions: {}", Utils.join(assignedPartitions, ", ")); @@ -351,7 +363,7 @@ protected void onJoinComplete(int generation, // should at least encode the short version if (assignmentBuffer.remaining() < 2) - throw new IllegalStateException("There is insufficient bytes available to read assignment from the sync-group response (" + + throw new IllegalStateException("There are insufficient bytes available to read assignment from the sync-group response (" + "actual byte size " + assignmentBuffer.remaining() + ") , this is not expected; " + "it is possible that the leader's assign function is buggy and did not return any assignment for this member, " + "or because static member is configured and the protocol is buggy hence did not get the assignment for this member"); @@ -406,11 +418,7 @@ protected void onJoinComplete(int generation, maybeUpdateJoinedSubscription(assignedPartitions); // Catch any exception here to make sure we could complete the user callback. - try { - assignor.onAssignment(assignment, groupMetadata); - } catch (Exception e) { - firstException.compareAndSet(null, e); - } + firstException.compareAndSet(null, invokeOnAssignment(assignor, assignment)); // Reschedule the auto commit starting from now if (autoCommitEnabled) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java index 4d19ef4a0141e..e5dd8a4a130b2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java @@ -32,6 +32,7 @@ public final class Heartbeat { private final Timer pollTimer; private volatile long lastHeartbeatSend = 0L; + private volatile boolean heartbeatInFlight = false; public Heartbeat(GroupRebalanceConfig config, Time time) { @@ -56,60 +57,68 @@ public void poll(long now) { pollTimer.reset(maxPollIntervalMs); } - public void sentHeartbeat(long now) { - this.lastHeartbeatSend = now; + boolean hasInflight() { + return heartbeatInFlight; + } + + void sentHeartbeat(long now) { + lastHeartbeatSend = now; + heartbeatInFlight = true; update(now); heartbeatTimer.reset(rebalanceConfig.heartbeatIntervalMs); } - public void failHeartbeat() { + + void failHeartbeat() { update(time.milliseconds()); + heartbeatInFlight = false; heartbeatTimer.reset(rebalanceConfig.retryBackoffMs); } - public void receiveHeartbeat() { + + void receiveHeartbeat() { update(time.milliseconds()); + heartbeatInFlight = false; sessionTimer.reset(rebalanceConfig.sessionTimeoutMs); } - public boolean shouldHeartbeat(long now) { + boolean shouldHeartbeat(long now) { update(now); return heartbeatTimer.isExpired(); } - public long lastHeartbeatSend() { + long lastHeartbeatSend() { return this.lastHeartbeatSend; } - public long timeToNextHeartbeat(long now) { + long timeToNextHeartbeat(long now) { update(now); return heartbeatTimer.remainingMs(); } - public boolean sessionTimeoutExpired(long now) { + boolean sessionTimeoutExpired(long now) { update(now); return sessionTimer.isExpired(); } - public void resetTimeouts() { + void resetTimeouts() { update(time.milliseconds()); sessionTimer.reset(rebalanceConfig.sessionTimeoutMs); pollTimer.reset(maxPollIntervalMs); heartbeatTimer.reset(rebalanceConfig.heartbeatIntervalMs); } - public void resetSessionTimeout() { + void resetSessionTimeout() { update(time.milliseconds()); sessionTimer.reset(rebalanceConfig.sessionTimeoutMs); } - public boolean pollTimeoutExpired(long now) { + boolean pollTimeoutExpired(long now) { update(now); return pollTimer.isExpired(); } - public long lastPollTime() { + long lastPollTime() { return pollTimer.currentTimeMs(); } - -} \ No newline at end of file +} diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index e2315cd478c3c..eb88e3a31e0a3 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -465,6 +465,52 @@ public void testSyncGroupRequestWithFencedInstanceIdException() { assertThrows(FencedInstanceIdException.class, () -> coordinator.ensureActiveGroup()); } + @Test + public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws InterruptedException { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, + "The heartbeat request was not sent in time after 2000ms elapsed"); + + assertTrue(coordinator.heartbeat().hasInflight()); + + // set the client to re-join group + mockClient.respond(heartbeatResponse(Errors.UNKNOWN_MEMBER_ID)); + + coordinator.requestRejoin(); + + TestUtils.waitForCondition(() -> { + coordinator.ensureActiveGroup(new MockTime(1L).timer(100L)); + return !coordinator.heartbeat().hasInflight(); + }, + 2000, + "The heartbeat response was not been received in time after 2000ms elapsed"); + + assertFalse(coordinator.heartbeat().hasInflight()); + + // the generation should be reset but the rebalance should still proceed + assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation()); + + mockClient.respond(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + assertEquals(currGen, coordinator.generation()); + } + @Test public void testHeartbeatRequestWithFencedInstanceIdException() throws InterruptedException { setupCoordinator(); diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index 0b3390e1d3139..f9928660ba149 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -173,7 +173,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { @Test def testMaxPollIntervalMs(): Unit = { - this.consumerConfig.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 3000.toString) + this.consumerConfig.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 1000.toString) this.consumerConfig.setProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 500.toString) this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 2000.toString) @@ -187,7 +187,9 @@ class PlaintextConsumerTest extends BaseConsumerTest { assertEquals(1, listener.callsToAssigned) assertEquals(0, listener.callsToRevoked) - Thread.sleep(3500) + // after we extend longer than max.poll a rebalance should be triggered + // NOTE we need to have a relatively much larger value than max.poll to let heartbeat expired for sure + Thread.sleep(3000) // we should fall out of the group and need to rebalance awaitRebalance(consumer, listener) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala index 55f0d1e1bf458..be080da3fe77f 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala @@ -616,10 +616,10 @@ class GroupCoordinatorTest { groupId, CompletingRebalance, Some(protocolType)) - assertEquals(leaderJoinGroupResult.leaderId, leaderJoinGroupResult.memberId) + assertEquals(rebalanceResult.leaderId, leaderJoinGroupResult.memberId) assertEquals(rebalanceResult.leaderId, leaderJoinGroupResult.leaderId) - // Old member shall be getting a successful join group response. + // Old follower shall be getting a successful join group response. val oldFollowerJoinGroupResult = Await.result(oldFollowerJoinGroupFuture, Duration(1, TimeUnit.MILLISECONDS)) checkJoinGroupResult(oldFollowerJoinGroupResult, Errors.NONE, @@ -629,31 +629,39 @@ class GroupCoordinatorTest { CompletingRebalance, Some(protocolType), expectedLeaderId = leaderJoinGroupResult.memberId) + assertEquals(rebalanceResult.followerId, oldFollowerJoinGroupResult.memberId) + assertEquals(rebalanceResult.leaderId, oldFollowerJoinGroupResult.leaderId) + assertTrue(getGroup(groupId).is(CompletingRebalance)) + // Duplicate follower joins group with unknown member id will trigger member.id replacement, + // and will also trigger a rebalance under CompletingRebalance state; the old follower sync callback + // will return fenced exception while broker replaces the member identity with the duplicate follower joins. EasyMock.reset(replicaManager) val oldFollowerSyncGroupFuture = sendSyncGroupFollower(groupId, oldFollowerJoinGroupResult.generationId, oldFollowerJoinGroupResult.memberId, Some(protocolType), Some(protocolName), followerInstanceId) - // Duplicate follower joins group with unknown member id will trigger member.id replacement. EasyMock.reset(replicaManager) val duplicateFollowerJoinFuture = sendJoinGroup(groupId, JoinGroupRequest.UNKNOWN_MEMBER_ID, protocolType, protocols, groupInstanceId = followerInstanceId) timer.advanceClock(1) - // Old follower sync callback will return fenced exception while broker replaces the member identity. val oldFollowerSyncGroupResult = Await.result(oldFollowerSyncGroupFuture, Duration(1, TimeUnit.MILLISECONDS)) assertEquals(Errors.FENCED_INSTANCE_ID, oldFollowerSyncGroupResult.error) + assertTrue(getGroup(groupId).is(PreparingRebalance)) + + timer.advanceClock(GroupInitialRebalanceDelay + 1) + timer.advanceClock(DefaultRebalanceTimeout + 1) - // Duplicate follower will get the same response as old follower. val duplicateFollowerJoinGroupResult = Await.result(duplicateFollowerJoinFuture, Duration(1, TimeUnit.MILLISECONDS)) checkJoinGroupResult(duplicateFollowerJoinGroupResult, Errors.NONE, - rebalanceResult.generation + 1, - Set.empty, + rebalanceResult.generation + 2, + Set(followerInstanceId), // this follower will become the new leader, and hence it would have the member list groupId, CompletingRebalance, Some(protocolType), - expectedLeaderId = leaderJoinGroupResult.memberId) + expectedLeaderId = duplicateFollowerJoinGroupResult.memberId) + assertTrue(getGroup(groupId).is(CompletingRebalance)) } @Test From 1e1ffb000c3108bed4af134edd14dc34e8e51874 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 7 Apr 2020 13:49:44 -0700 Subject: [PATCH 3/9] remember generation --- .../internals/AbstractCoordinator.java | 107 +++++++++++++----- .../internals/ConsumerCoordinator.java | 37 ++++-- .../common/requests/LeaveGroupResponse.java | 5 + .../common/requests/SyncGroupResponse.java | 5 + .../internals/AbstractCoordinatorTest.java | 92 +++++++++++++++ 5 files changed, 212 insertions(+), 34 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index 4844922c64afe..2d185053f3b23 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -568,10 +568,14 @@ RequestFuture sendJoinGroupRequest() { int joinGroupTimeoutMs = Math.max(rebalanceConfig.rebalanceTimeoutMs, rebalanceConfig.rebalanceTimeoutMs + 5000); return client.send(coordinator, requestBuilder, joinGroupTimeoutMs) - .compose(new JoinGroupResponseHandler()); + .compose(new JoinGroupResponseHandler(generation)); } private class JoinGroupResponseHandler extends CoordinatorResponseHandler { + private JoinGroupResponseHandler(final Generation generation) { + super(generation); + } + @Override public void handle(JoinGroupResponse joinResponse, RequestFuture future) { Errors error = joinResponse.error(); @@ -606,9 +610,12 @@ public void handle(JoinGroupResponse joinResponse, RequestFuture fut // backoff and retry future.raise(error); } else if (error == Errors.UNKNOWN_MEMBER_ID) { - // reset the member id and retry immediately - resetGenerationOnResponseError(ApiKeys.JOIN_GROUP, error); - log.debug("Attempt to join group failed due to unknown member id."); + log.debug("Attempt to join group failed due to unknown member id with {}.", sentGeneration); + // only need to reset the member id if generation has not been changed, + // then retry immediately + if (generationUnchanged()) + resetGenerationOnResponseError(ApiKeys.JOIN_GROUP, error); + future.raise(error); } else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) { @@ -617,7 +624,10 @@ public void handle(JoinGroupResponse joinResponse, RequestFuture fut log.debug("Attempt to join group failed due to obsolete coordinator information: {}", error.message()); future.raise(error); } else if (error == Errors.FENCED_INSTANCE_ID) { - log.error("Received fatal exception: group.instance.id gets fenced"); + // for join-group request, even if the generation has changed we would not expect the instance id + // gets fenced, and hence we always treat this as a fatal error + log.error("Attempt to join group failed due to group instance id {} gets fenced with {}", + rebalanceConfig.groupInstanceId, sentGeneration); future.raise(error); } else if (error == Errors.INCONSISTENT_GROUP_PROTOCOL || error == Errors.INVALID_SESSION_TIMEOUT @@ -708,10 +718,14 @@ private RequestFuture sendSyncGroupRequest(SyncGroupRequest.Builder if (coordinatorUnknown()) return RequestFuture.coordinatorNotAvailable(); return client.send(coordinator, requestBuilder) - .compose(new SyncGroupResponseHandler()); + .compose(new SyncGroupResponseHandler(generation)); } private class SyncGroupResponseHandler extends CoordinatorResponseHandler { + private SyncGroupResponseHandler(final Generation generation) { + super(generation); + } + @Override public void handle(SyncGroupResponse syncResponse, RequestFuture future) { @@ -737,16 +751,21 @@ public void handle(SyncGroupResponse syncResponse, log.debug("SyncGroup failed because the group began another rebalance"); future.raise(error); } else if (error == Errors.FENCED_INSTANCE_ID) { - log.error("Received fatal exception: group.instance.id gets fenced"); + // for sync-group request, even if the generation has changed we would not expect the instance id + // gets fenced, and hence we always treat this as a fatal error + log.error("SyncGroup failed with {} due to group.instance.id {} gets fenced", + sentGeneration, rebalanceConfig.groupInstanceId); future.raise(error); } else if (error == Errors.UNKNOWN_MEMBER_ID || error == Errors.ILLEGAL_GENERATION) { - log.debug("SyncGroup failed: {}", error.message()); - resetGenerationOnResponseError(ApiKeys.SYNC_GROUP, error); + log.info("SyncGroup failed with {}: {}, would request re-join", sentGeneration, error.message()); + if (generationUnchanged()) + resetGenerationOnResponseError(ApiKeys.SYNC_GROUP, error); + future.raise(error); } else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) { - log.debug("SyncGroup failed: {}", error.message()); + log.debug("SyncGroup failed: {}, marking coordinator unknown", error.message()); markCoordinatorUnknown(); future.raise(error); } else { @@ -989,7 +1008,7 @@ public synchronized RequestFuture maybeLeaveGroup(String leaveReason) { Collections.singletonList(new MemberIdentity().setMemberId(generation.memberId)) ); - future = client.send(coordinator, request).compose(new LeaveGroupResponseHandler()); + future = client.send(coordinator, request).compose(new LeaveGroupResponseHandler(generation)); client.pollNoWakeup(); } @@ -1003,6 +1022,10 @@ protected boolean isDynamicMember() { } private class LeaveGroupResponseHandler extends CoordinatorResponseHandler { + private LeaveGroupResponseHandler(final Generation generation) { + super(generation); + } + @Override public void handle(LeaveGroupResponse leaveResponse, RequestFuture future) { final List members = leaveResponse.memberResponses(); @@ -1013,10 +1036,10 @@ public void handle(LeaveGroupResponse leaveResponse, RequestFuture future) final Errors error = leaveResponse.error(); if (error == Errors.NONE) { - log.debug("LeaveGroup request returned successfully"); + log.debug("LeaveGroup response with {} returned successfully: {}", sentGeneration, response); future.complete(null); } else { - log.error("LeaveGroup request failed with error: {}", error.message()); + log.error("LeaveGroup response with {} failed with error: {}", sentGeneration, error.message()); future.raise(error); } } @@ -1037,10 +1060,8 @@ synchronized RequestFuture sendHeartbeatRequest() { } private class HeartbeatResponseHandler extends CoordinatorResponseHandler { - private final Generation sentGeneration; - private HeartbeatResponseHandler(final Generation generation) { - this.sentGeneration = generation; + super(generation); } @Override @@ -1061,16 +1082,36 @@ public void handle(HeartbeatResponse heartbeatResponse, RequestFuture futu requestRejoin(); future.raise(error); } else if (error == Errors.ILLEGAL_GENERATION) { - log.info("Attempt to heartbeat failed since generation {} is not current", sentGeneration.generationId); - resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); - future.raise(error); + if (generationUnchanged()) { + log.info("Attempt to heartbeat failed since current {} is not valid, resetting generation", sentGeneration); + resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); + future.raise(error); + } else { + // if the generation has changed, then ignore this error + log.info("Attempt to heartbeat failed since old {} is not valid, ignoring the error", sentGeneration); + future.complete(null); + } } else if (error == Errors.FENCED_INSTANCE_ID) { - log.error("Received fatal exception: group.instance.id gets fenced"); - future.raise(error); + if (generationUnchanged()) { + log.info("Attempt to heartbeat failed since current {} gets fenced with group instance id {}", + sentGeneration, rebalanceConfig.groupInstanceId); + future.raise(error); + } else { + // if the generation has changed, then ignore this error + log.info("Attempt to heartbeat failed since old {} gets fenced with group instance id {}, " + + "ignoring the error", sentGeneration, rebalanceConfig.groupInstanceId); + future.complete(null); + } } else if (error == Errors.UNKNOWN_MEMBER_ID) { - log.info("Attempt to heartbeat failed since member id {} is not valid.", sentGeneration.memberId); - resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); - future.raise(error); + if (generationUnchanged()) { + log.info("Attempt to heartbeat failed since current {} member id is unknown, resetting generation", sentGeneration); + resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); + future.raise(error); + } else { + // if the generation has changed, then ignore this error + log.info("Attempt to heartbeat failed since old {} member id is unknown, ignoring the error", sentGeneration); + future.complete(null); + } } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) { future.raise(GroupAuthorizationException.forGroupId(rebalanceConfig.groupId)); } else { @@ -1080,7 +1121,12 @@ public void handle(HeartbeatResponse heartbeatResponse, RequestFuture futu } protected abstract class CoordinatorResponseHandler extends RequestFutureAdapter { - protected ClientResponse response; + CoordinatorResponseHandler(final Generation generation) { + this.sentGeneration = generation; + } + + final Generation sentGeneration; + ClientResponse response; public abstract void handle(R response, RequestFuture future); @@ -1106,6 +1152,11 @@ public void onSuccess(ClientResponse clientResponse, RequestFuture future) { } } + boolean generationUnchanged() { + synchronized (AbstractCoordinator.this) { + return generation.equals(sentGeneration); + } + } } protected Meter createMeter(Metrics metrics, String groupName, String baseName, String descriptiveName) { @@ -1421,7 +1472,7 @@ private static class UnjoinedGroupException extends RetriableException { } // For testing only below - public Heartbeat heartbeat() { + final Heartbeat heartbeat() { return heartbeat; } @@ -1449,4 +1500,8 @@ final boolean hasUnknownGeneration() { final boolean hasValidMemberId() { return generation != Generation.NO_GENERATION && generation.hasMemberId(); } + + final void setNewGeneration(final Generation generation) { + this.generation = generation; + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index b2bb2345e0459..675c11b4f1a7e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -1143,14 +1143,14 @@ private RequestFuture sendOffsetCommitRequest(final Map { - private final Map offsets; - private OffsetCommitResponseHandler(Map offsets) { + private OffsetCommitResponseHandler(Map offsets, Generation generation) { + super(generation); this.offsets = offsets; } @@ -1198,8 +1198,16 @@ public void handle(OffsetCommitResponse commitResponse, RequestFuture futu future.raise(error); return; } else if (error == Errors.FENCED_INSTANCE_ID) { - log.error("Received fatal exception: group.instance.id gets fenced"); - future.raise(error); + log.info("OffsetCommit failed with {} due to group instance id {} fenced", sentGeneration, rebalanceConfig.groupInstanceId); + + // if the generation has changed, do not raise the fatal error but rebalance-in-progress + if (generationUnchanged()) { + future.raise(error); + } else { + future.raise(new RebalanceInProgressException("Offset commit cannot be completed since the " + + "consumer member's old generation is fenced by its group instance id, it is possible that " + + "this consumer has already participated another rebalance and got a new generation")); + } return; } else if (error == Errors.REBALANCE_IN_PROGRESS) { /* Consumer should not try to commit offset in between join-group and sync-group, @@ -1217,9 +1225,18 @@ public void handle(OffsetCommitResponse commitResponse, RequestFuture futu return; } else if (error == Errors.UNKNOWN_MEMBER_ID || error == Errors.ILLEGAL_GENERATION) { - // need to reset generation and re-join group - resetGenerationOnResponseError(ApiKeys.OFFSET_COMMIT, error); - future.raise(new CommitFailedException()); + log.info("OffsetCommit failed with {}: {}", sentGeneration, error.message()); + + // only need to reset generation and re-join group if generation has not changed; + // otherwise only raise rebalance-in-progress error + if (generationUnchanged()) { + resetGenerationOnResponseError(ApiKeys.OFFSET_COMMIT, error); + future.raise(new CommitFailedException()); + } else { + future.raise(new RebalanceInProgressException("Offset commit cannot be completed since the " + + "consumer member's generation is already stale, meaning it has already participated another rebalance and " + + "got a new generation. You can try completing the rebalance by calling poll() and then retry commit again")); + } return; } else { future.raise(new KafkaException("Unexpected error in commit: " + error.message())); @@ -1261,6 +1278,10 @@ private RequestFuture> sendOffsetFetchReq } private class OffsetFetchResponseHandler extends CoordinatorResponseHandler> { + private OffsetFetchResponseHandler() { + super(Generation.NO_GENERATION); + } + @Override public void handle(OffsetFetchResponse response, RequestFuture> future) { if (response.hasError()) { diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java index e64fde168980b..cee54af5f24cd 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java @@ -134,6 +134,11 @@ public Map errorCounts() { return combinedErrorCounts; } + @Override + public String toString() { + return data.toString(); + } + @Override public Struct toStruct(short version) { return data.toStruct(version); diff --git a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java index 51765e8f51ff9..f627bb774a688 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java @@ -61,6 +61,11 @@ protected Struct toStruct(short version) { return data.toStruct(version); } + @Override + public String toString() { + return data.toString(); + } + public static SyncGroupResponse parse(ByteBuffer buffer, short version) { return new SyncGroupResponse(ApiKeys.SYNC_GROUP.parseResponse(version, buffer), version); } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index eb88e3a31e0a3..94b9cc6eda380 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -465,6 +465,98 @@ public void testSyncGroupRequestWithFencedInstanceIdException() { assertThrows(FencedInstanceIdException.class, () -> coordinator.ensureActiveGroup()); } + @Test + public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + long startMs = System.currentTimeMillis(); + while (System.currentTimeMillis() - startMs < 1000 && !coordinator.heartbeat().hasInflight()) { + Thread.sleep(10); + } + + assertTrue(coordinator.heartbeat().hasInflight()); + + // change the generation + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + currGen.generationId + 1, + currGen.memberId, + currGen.protocolName); + coordinator.setNewGeneration(newGen); + + mockClient.respond(heartbeatResponse(Errors.ILLEGAL_GENERATION)); + + // the heartbeat error code should be ignored + startMs = System.currentTimeMillis(); + while (System.currentTimeMillis() - startMs < 1000 && coordinator.heartbeat().hasInflight()) { + Thread.sleep(10); + coordinator.pollHeartbeat(mockTime.milliseconds()); + } + + assertFalse(coordinator.heartbeat().hasInflight()); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + long startMs = System.currentTimeMillis(); + while (System.currentTimeMillis() - startMs < 1000 && !coordinator.heartbeat().hasInflight()) { + Thread.sleep(10); + } + + assertTrue(coordinator.heartbeat().hasInflight()); + + // change the generation + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + currGen.generationId, + currGen.memberId + "-new", + currGen.protocolName); + coordinator.setNewGeneration(newGen); + + mockClient.respond(heartbeatResponse(Errors.UNKNOWN_MEMBER_ID)); + + // the heartbeat error code should be ignored + startMs = System.currentTimeMillis(); + while (System.currentTimeMillis() - startMs < 1000 && coordinator.heartbeat().hasInflight()) { + Thread.sleep(10); + coordinator.pollHeartbeat(mockTime.milliseconds()); + } + + assertFalse(coordinator.heartbeat().hasInflight()); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + @Test public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws InterruptedException { setupCoordinator(); From ea6dbfc3e026f2ebcd9e1708c6133f1b26dc70e7 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 7 Apr 2020 22:25:58 -0700 Subject: [PATCH 4/9] add unit tests --- .../internals/AbstractCoordinator.java | 4 + .../internals/ConsumerCoordinator.java | 4 +- .../internals/AbstractCoordinatorTest.java | 184 +++++++++++++++--- .../internals/ConsumerCoordinatorTest.java | 87 +++++++++ 4 files changed, 250 insertions(+), 29 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index c37437cb320d6..844bdecf0ddd3 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -1504,4 +1504,8 @@ final boolean hasValidMemberId() { final void setNewGeneration(final Generation generation) { this.generation = generation; } + + final void setNewState(final MemberState state) { + this.state = state; + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index 675c11b4f1a7e..208193fb55339 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -1072,10 +1072,12 @@ public void onComplete(Map offsets, Exception * which returns a request future that can be polled in the case of a synchronous commit or ignored in the * asynchronous case. * + * NOTE: This is visible only for testing + * * @param offsets The list of offsets per partition that should be committed. * @return A request future whose value indicates whether the commit was successful or not */ - private RequestFuture sendOffsetCommitRequest(final Map offsets) { + RequestFuture sendOffsetCommitRequest(final Map offsets) { if (offsets.isEmpty()) return RequestFuture.voidSuccess(); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index 94b9cc6eda380..add55b382033f 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -465,6 +465,108 @@ public void testSyncGroupRequestWithFencedInstanceIdException() { assertThrows(FencedInstanceIdException.class, () -> coordinator.ensureActiveGroup()); } + @Test + public void testJoinGroupUnknownMemberResponseWithOldGeneration() { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + RequestFuture future = coordinator.sendJoinGroupRequest(); + + // change the generation after the join-group request + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + currGen.generationId, + currGen.memberId + "-new", + currGen.protocolName); + coordinator.setNewGeneration(newGen); + + mockClient.prepareResponse(joinGroupFollowerResponse(generation + 1, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.UNKNOWN_MEMBER_ID)); + + assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); + assertTrue(future.exception().getClass().isInstance(Errors.UNKNOWN_MEMBER_ID.exception())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testSyncGroupUnknownMemberResponseWithOldGeneration() { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + coordinator.ensureCoordinatorReady(mockTime.timer(0)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + RequestFuture future = coordinator.sendJoinGroupRequest(); + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + + // change the generation after the sync-group request + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + currGen.generationId, + currGen.memberId + "-new", + currGen.protocolName); + coordinator.setNewGeneration(newGen); + + mockClient.prepareResponse(syncGroupResponse(Errors.UNKNOWN_MEMBER_ID)); + assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); + assertTrue(future.exception().getClass().isInstance(Errors.UNKNOWN_MEMBER_ID.exception())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testSyncGroupIllegalGenerationResponseWithOldGeneration() { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + coordinator.ensureCoordinatorReady(mockTime.timer(0)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + RequestFuture future = coordinator.sendJoinGroupRequest(); + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + + // change the generation after the sync-group request + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + currGen.generationId + 1, + currGen.memberId, + currGen.protocolName); + coordinator.setNewGeneration(newGen); + + mockClient.prepareResponse(syncGroupResponse(Errors.ILLEGAL_GENERATION)); + assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); + assertTrue(future.exception().getClass().isInstance(Errors.ILLEGAL_GENERATION.exception())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + @Test public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); @@ -482,12 +584,8 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int // let the heartbeat request to send out a request mockTime.sleep(HEARTBEAT_INTERVAL_MS); - long startMs = System.currentTimeMillis(); - while (System.currentTimeMillis() - startMs < 1000 && !coordinator.heartbeat().hasInflight()) { - Thread.sleep(10); - } - - assertTrue(coordinator.heartbeat().hasInflight()); + TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, + "The heartbeat request was not sent in time after 2000ms elapsed"); // change the generation final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -499,13 +597,11 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int mockClient.respond(heartbeatResponse(Errors.ILLEGAL_GENERATION)); // the heartbeat error code should be ignored - startMs = System.currentTimeMillis(); - while (System.currentTimeMillis() - startMs < 1000 && coordinator.heartbeat().hasInflight()) { - Thread.sleep(10); - coordinator.pollHeartbeat(mockTime.milliseconds()); - } - - assertFalse(coordinator.heartbeat().hasInflight()); + TestUtils.waitForCondition(() -> { + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, + "The heartbeat response was not received in time after 2000ms elapsed"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); @@ -528,12 +624,8 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru // let the heartbeat request to send out a request mockTime.sleep(HEARTBEAT_INTERVAL_MS); - long startMs = System.currentTimeMillis(); - while (System.currentTimeMillis() - startMs < 1000 && !coordinator.heartbeat().hasInflight()) { - Thread.sleep(10); - } - - assertTrue(coordinator.heartbeat().hasInflight()); + TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, + "The heartbeat request was not sent in time after 2000ms elapsed"); // change the generation final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -545,13 +637,11 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru mockClient.respond(heartbeatResponse(Errors.UNKNOWN_MEMBER_ID)); // the heartbeat error code should be ignored - startMs = System.currentTimeMillis(); - while (System.currentTimeMillis() - startMs < 1000 && coordinator.heartbeat().hasInflight()) { - Thread.sleep(10); + TestUtils.waitForCondition(() -> { coordinator.pollHeartbeat(mockTime.milliseconds()); - } - - assertFalse(coordinator.heartbeat().hasInflight()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, + "The heartbeat response was not received in time after 2000ms elapsed"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); @@ -589,9 +679,7 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not been received in time after 2000ms elapsed"); - - assertFalse(coordinator.heartbeat().hasInflight()); + "The heartbeat response was not received in time after 2000ms elapsed"); // the generation should be reset but the rebalance should still proceed assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation()); @@ -603,6 +691,46 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru assertEquals(currGen, coordinator.generation()); } + @Test + public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, + "The heartbeat request was not sent in time after 2000ms elapsed"); + + // change the generation + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + currGen.generationId, + currGen.memberId + "-new", + currGen.protocolName); + coordinator.setNewGeneration(newGen); + + mockClient.respond(heartbeatResponse(Errors.FENCED_INSTANCE_ID)); + + // the heartbeat error code should be ignored + TestUtils.waitForCondition(() -> { + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, + "The heartbeat response was not received in time after 2000ms elapsed"); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + @Test public void testHeartbeatRequestWithFencedInstanceIdException() throws InterruptedException { setupCoordinator(); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java index aaf29625955ba..c62d59a8a09a9 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java @@ -2069,6 +2069,93 @@ public void testCommitOffsetUnknownMemberId() { new OffsetAndMetadata(100L, "metadata")), time.timer(Long.MAX_VALUE)); } + @Test + public void testCommitOffsetIllegalGenerationWithNewGenearion() { + client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE)); + + final AbstractCoordinator.Generation currGen = new AbstractCoordinator.Generation( + 1, + "memberId", + null); + coordinator.setNewGeneration(currGen); + + prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.ILLEGAL_GENERATION); + RequestFuture future = coordinator.sendOffsetCommitRequest(singletonMap(t1p, + new OffsetAndMetadata(100L, "metadata"))); + + // change the generation + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + 2, + "memberId-new", + null); + coordinator.setNewGeneration(newGen); + + assertTrue(consumerClient.poll(future, time.timer(30000))); + assertTrue(future.exception().getClass().isInstance(Errors.REBALANCE_IN_PROGRESS.exception())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testCommitOffsetUnknownMemberWithNewGenearion() { + client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE)); + + final AbstractCoordinator.Generation currGen = new AbstractCoordinator.Generation( + 1, + "memberId", + null); + coordinator.setNewGeneration(currGen); + + prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.UNKNOWN_MEMBER_ID); + RequestFuture future = coordinator.sendOffsetCommitRequest(singletonMap(t1p, + new OffsetAndMetadata(100L, "metadata"))); + + // change the generation + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + 2, + "memberId-new", + null); + coordinator.setNewGeneration(newGen); + + assertTrue(consumerClient.poll(future, time.timer(30000))); + assertTrue(future.exception().getClass().isInstance(Errors.REBALANCE_IN_PROGRESS.exception())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testCommitOffsetFencedInstanceWithNewGenearion() { + client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE)); + + final AbstractCoordinator.Generation currGen = new AbstractCoordinator.Generation( + 1, + "memberId", + null); + coordinator.setNewGeneration(currGen); + + prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.FENCED_INSTANCE_ID); + RequestFuture future = coordinator.sendOffsetCommitRequest(singletonMap(t1p, + new OffsetAndMetadata(100L, "metadata"))); + + // change the generation + final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( + 2, + "memberId-new", + null); + coordinator.setNewGeneration(newGen); + + assertTrue(consumerClient.poll(future, time.timer(30000))); + assertTrue(future.exception().getClass().isInstance(Errors.REBALANCE_IN_PROGRESS.exception())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + @Test public void testCommitOffsetRebalanceInProgress() { // we cannot retry if a rebalance occurs before the commit completed From 69de4bd2ebf46ce9d60eee7c56dc6345ebf8ebff Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 7 Apr 2020 22:31:52 -0700 Subject: [PATCH 5/9] checkstyle fixes --- .../consumer/internals/AbstractCoordinatorTest.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index add55b382033f..6d15ed74e9814 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -598,9 +598,9 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int // the heartbeat error code should be ignored TestUtils.waitForCondition(() -> { - coordinator.pollHeartbeat(mockTime.milliseconds()); - return !coordinator.heartbeat().hasInflight(); - }, 2000, + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, "The heartbeat response was not received in time after 2000ms elapsed"); // the generation should not be reset @@ -722,9 +722,9 @@ public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws Interr // the heartbeat error code should be ignored TestUtils.waitForCondition(() -> { - coordinator.pollHeartbeat(mockTime.milliseconds()); - return !coordinator.heartbeat().hasInflight(); - }, 2000, + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, "The heartbeat response was not received in time after 2000ms elapsed"); // the generation should not be reset From c37e9f7aebc655350052ec1e1ac940e6ec018d12 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Wed, 15 Apr 2020 12:03:05 -0700 Subject: [PATCH 6/9] github comments --- .../internals/AbstractCoordinator.java | 31 ++----- .../internals/AbstractCoordinatorTest.java | 84 +++++-------------- 2 files changed, 29 insertions(+), 86 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index 844bdecf0ddd3..f06e24afa7abf 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -1081,35 +1081,18 @@ public void handle(HeartbeatResponse heartbeatResponse, RequestFuture futu log.info("Attempt to heartbeat failed since group is rebalancing"); requestRejoin(); future.raise(error); - } else if (error == Errors.ILLEGAL_GENERATION) { + } else if (error == Errors.ILLEGAL_GENERATION || + error == Errors.UNKNOWN_MEMBER_ID || + error == Errors.FENCED_INSTANCE_ID) { if (generationUnchanged()) { - log.info("Attempt to heartbeat failed since current {} is not valid, resetting generation", sentGeneration); + log.info("Attempt to heartbeat failed with generation {} and group instance id {} due to {}, resetting generation", + sentGeneration, rebalanceConfig.groupInstanceId, error.message()); resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); future.raise(error); } else { // if the generation has changed, then ignore this error - log.info("Attempt to heartbeat failed since old {} is not valid, ignoring the error", sentGeneration); - future.complete(null); - } - } else if (error == Errors.FENCED_INSTANCE_ID) { - if (generationUnchanged()) { - log.info("Attempt to heartbeat failed since current {} gets fenced with group instance id {}", - sentGeneration, rebalanceConfig.groupInstanceId); - future.raise(error); - } else { - // if the generation has changed, then ignore this error - log.info("Attempt to heartbeat failed since old {} gets fenced with group instance id {}, " + - "ignoring the error", sentGeneration, rebalanceConfig.groupInstanceId); - future.complete(null); - } - } else if (error == Errors.UNKNOWN_MEMBER_ID) { - if (generationUnchanged()) { - log.info("Attempt to heartbeat failed since current {} member id is unknown, resetting generation", sentGeneration); - resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); - future.raise(error); - } else { - // if the generation has changed, then ignore this error - log.info("Attempt to heartbeat failed since old {} member id is unknown, ignoring the error", sentGeneration); + log.info("Attempt to heartbeat failed with stale generation {} and group instance id {} due to {}, ignoring the error", + sentGeneration, rebalanceConfig.groupInstanceId, error.message()); future.complete(null); } } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) { diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index 6d15ed74e9814..3c01bc763359c 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -144,6 +144,17 @@ false, false, new SubscriptionState(logContext, OffsetResetStrategy.EARLIEST), mockTime); } + private void joinGroup() { + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + + final int generation = 1; + + mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + } + @Test public void testMetrics() { setupCoordinator(); @@ -468,14 +479,7 @@ public void testSyncGroupRequestWithFencedInstanceIdException() { @Test public void testJoinGroupUnknownMemberResponseWithOldGeneration() { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); @@ -488,7 +492,7 @@ public void testJoinGroupUnknownMemberResponseWithOldGeneration() { currGen.protocolName); coordinator.setNewGeneration(newGen); - mockClient.prepareResponse(joinGroupFollowerResponse(generation + 1, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.UNKNOWN_MEMBER_ID)); + mockClient.prepareResponse(joinGroupFollowerResponse(currGen.generationId + 1, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.UNKNOWN_MEMBER_ID)); assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); assertTrue(future.exception().getClass().isInstance(Errors.UNKNOWN_MEMBER_ID.exception())); @@ -500,20 +504,12 @@ public void testJoinGroupUnknownMemberResponseWithOldGeneration() { @Test public void testSyncGroupUnknownMemberResponseWithOldGeneration() { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - coordinator.ensureCoordinatorReady(mockTime.timer(0)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); RequestFuture future = coordinator.sendJoinGroupRequest(); consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); @@ -535,20 +531,12 @@ public void testSyncGroupUnknownMemberResponseWithOldGeneration() { @Test public void testSyncGroupIllegalGenerationResponseWithOldGeneration() { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - coordinator.ensureCoordinatorReady(mockTime.timer(0)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.prepareResponse(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); RequestFuture future = coordinator.sendJoinGroupRequest(); consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); @@ -570,14 +558,7 @@ public void testSyncGroupIllegalGenerationResponseWithOldGeneration() { @Test public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); @@ -610,14 +591,7 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int @Test public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); @@ -650,14 +624,7 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru @Test public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws InterruptedException { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); @@ -684,7 +651,7 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru // the generation should be reset but the rebalance should still proceed assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation()); - mockClient.respond(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + mockClient.respond(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); coordinator.ensureActiveGroup(); @@ -694,14 +661,7 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru @Test public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); - - final int generation = 1; - - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); - - coordinator.ensureActiveGroup(); + joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); From d04ae3b1a5fa2a7f53f5b888b7e6f33e57237617 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Wed, 15 Apr 2020 15:29:19 -0700 Subject: [PATCH 7/9] minor refacotring on unit tests and github comments --- .../internals/AbstractCoordinator.java | 6 +- .../internals/AbstractCoordinatorTest.java | 80 +++++++++++++------ 2 files changed, 59 insertions(+), 27 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index f06e24afa7abf..9c2dfa140ab13 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -1459,7 +1459,7 @@ final Heartbeat heartbeat() { return heartbeat; } - final void setLastRebalanceTime(final long timestamp) { + final synchronized void setLastRebalanceTime(final long timestamp) { lastRebalanceEndMs = timestamp; } @@ -1484,11 +1484,11 @@ final boolean hasValidMemberId() { return generation != Generation.NO_GENERATION && generation.hasMemberId(); } - final void setNewGeneration(final Generation generation) { + final synchronized void setNewGeneration(final Generation generation) { this.generation = generation; } - final void setNewState(final MemberState state) { + final synchronized void setNewState(final MemberState state) { this.state = state; } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index 3c01bc763359c..3a27dd413a201 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -477,7 +477,7 @@ public void testSyncGroupRequestWithFencedInstanceIdException() { } @Test - public void testJoinGroupUnknownMemberResponseWithOldGeneration() { + public void testJoinGroupUnknownMemberResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); joinGroup(); @@ -485,6 +485,9 @@ public void testJoinGroupUnknownMemberResponseWithOldGeneration() { RequestFuture future = coordinator.sendJoinGroupRequest(); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The join-group request was not sent in time after"); + // change the generation after the join-group request final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( currGen.generationId, @@ -492,7 +495,7 @@ public void testJoinGroupUnknownMemberResponseWithOldGeneration() { currGen.protocolName); coordinator.setNewGeneration(newGen); - mockClient.prepareResponse(joinGroupFollowerResponse(currGen.generationId + 1, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.UNKNOWN_MEMBER_ID)); + mockClient.respond(joinGroupFollowerResponse(currGen.generationId + 1, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.UNKNOWN_MEMBER_ID)); assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); assertTrue(future.exception().getClass().isInstance(Errors.UNKNOWN_MEMBER_ID.exception())); @@ -502,16 +505,29 @@ public void testJoinGroupUnknownMemberResponseWithOldGeneration() { } @Test - public void testSyncGroupUnknownMemberResponseWithOldGeneration() { + public void testSyncGroupUnknownMemberResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); - mockClient.prepareResponse(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); RequestFuture future = coordinator.sendJoinGroupRequest(); - consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + + TestUtils.waitForCondition(() -> { + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + return !mockClient.requests().isEmpty(); + }, 2000, + "The join-group request was not sent in time"); + + mockClient.respond(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + assertTrue(mockClient.requests().isEmpty()); + + TestUtils.waitForCondition(() -> { + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + return !mockClient.requests().isEmpty(); + }, 2000, + "The sync-group request was not sent in time"); // change the generation after the sync-group request final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -520,7 +536,7 @@ public void testSyncGroupUnknownMemberResponseWithOldGeneration() { currGen.protocolName); coordinator.setNewGeneration(newGen); - mockClient.prepareResponse(syncGroupResponse(Errors.UNKNOWN_MEMBER_ID)); + mockClient.respond(syncGroupResponse(Errors.UNKNOWN_MEMBER_ID)); assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); assertTrue(future.exception().getClass().isInstance(Errors.UNKNOWN_MEMBER_ID.exception())); @@ -529,25 +545,38 @@ public void testSyncGroupUnknownMemberResponseWithOldGeneration() { } @Test - public void testSyncGroupIllegalGenerationResponseWithOldGeneration() { + public void testSyncGroupIllegalGenerationResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); joinGroup(); final AbstractCoordinator.Generation currGen = coordinator.generation(); coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); - mockClient.prepareResponse(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); RequestFuture future = coordinator.sendJoinGroupRequest(); - consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + + TestUtils.waitForCondition(() -> { + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + return !mockClient.requests().isEmpty(); + }, 2000, + "The join-group request was not sent in time"); + + mockClient.respond(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); + assertTrue(mockClient.requests().isEmpty()); + + TestUtils.waitForCondition(() -> { + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + return !mockClient.requests().isEmpty(); + }, 2000, + "The sync-group request was not sent in time"); // change the generation after the sync-group request final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( - currGen.generationId + 1, - currGen.memberId, + currGen.generationId, + currGen.memberId + "-new", currGen.protocolName); coordinator.setNewGeneration(newGen); - mockClient.prepareResponse(syncGroupResponse(Errors.ILLEGAL_GENERATION)); + mockClient.respond(syncGroupResponse(Errors.ILLEGAL_GENERATION)); assertTrue(consumerClient.poll(future, mockTime.timer(REQUEST_TIMEOUT_MS))); assertTrue(future.exception().getClass().isInstance(Errors.ILLEGAL_GENERATION.exception())); @@ -565,8 +594,9 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int // let the heartbeat request to send out a request mockTime.sleep(HEARTBEAT_INTERVAL_MS); - TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, - "The heartbeat request was not sent in time after 2000ms elapsed"); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent in time"); + assertTrue(coordinator.heartbeat().hasInflight()); // change the generation final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -582,7 +612,7 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int coordinator.pollHeartbeat(mockTime.milliseconds()); return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time after 2000ms elapsed"); + "The heartbeat response was not received in time"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); @@ -598,8 +628,9 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru // let the heartbeat request to send out a request mockTime.sleep(HEARTBEAT_INTERVAL_MS); - TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, - "The heartbeat request was not sent in time after 2000ms elapsed"); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent in time"); + assertTrue(coordinator.heartbeat().hasInflight()); // change the generation final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -615,7 +646,7 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru coordinator.pollHeartbeat(mockTime.milliseconds()); return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time after 2000ms elapsed"); + "The heartbeat response was not received in time"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); @@ -631,8 +662,8 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru // let the heartbeat request to send out a request mockTime.sleep(HEARTBEAT_INTERVAL_MS); - TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, - "The heartbeat request was not sent in time after 2000ms elapsed"); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent in time"); assertTrue(coordinator.heartbeat().hasInflight()); @@ -646,7 +677,7 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time after 2000ms elapsed"); + "The heartbeat response was not received in time"); // the generation should be reset but the rebalance should still proceed assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation()); @@ -668,8 +699,9 @@ public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws Interr // let the heartbeat request to send out a request mockTime.sleep(HEARTBEAT_INTERVAL_MS); - TestUtils.waitForCondition(() -> coordinator.heartbeat().hasInflight(), 2000, - "The heartbeat request was not sent in time after 2000ms elapsed"); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent in time"); + assertTrue(coordinator.heartbeat().hasInflight()); // change the generation final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -685,7 +717,7 @@ public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws Interr coordinator.pollHeartbeat(mockTime.milliseconds()); return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time after 2000ms elapsed"); + "The heartbeat response was not received in time"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); From 2793e1eee6d0018ed9858aa20094783456844a17 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Fri, 17 Apr 2020 11:11:11 -0700 Subject: [PATCH 8/9] github comments --- .../consumer/internals/AbstractCoordinator.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index 9c2dfa140ab13..9b4307f1dd95c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -626,7 +626,7 @@ public void handle(JoinGroupResponse joinResponse, RequestFuture fut } else if (error == Errors.FENCED_INSTANCE_ID) { // for join-group request, even if the generation has changed we would not expect the instance id // gets fenced, and hence we always treat this as a fatal error - log.error("Attempt to join group failed due to group instance id {} gets fenced with {}", + log.error("Attempt to join group with generation {} failed because the group instance id {} has been fenced by another instance", rebalanceConfig.groupInstanceId, sentGeneration); future.raise(error); } else if (error == Errors.INCONSISTENT_GROUP_PROTOCOL @@ -753,12 +753,12 @@ public void handle(SyncGroupResponse syncResponse, } else if (error == Errors.FENCED_INSTANCE_ID) { // for sync-group request, even if the generation has changed we would not expect the instance id // gets fenced, and hence we always treat this as a fatal error - log.error("SyncGroup failed with {} due to group.instance.id {} gets fenced", + log.error("SyncGroup with {} failed because the group instance id {} has been fenced by another instance", sentGeneration, rebalanceConfig.groupInstanceId); future.raise(error); } else if (error == Errors.UNKNOWN_MEMBER_ID || error == Errors.ILLEGAL_GENERATION) { - log.info("SyncGroup failed with {}: {}, would request re-join", sentGeneration, error.message()); + log.info("SyncGroup with {} failed: {}, would request re-join", sentGeneration, error.message()); if (generationUnchanged()) resetGenerationOnResponseError(ApiKeys.SYNC_GROUP, error); @@ -1039,7 +1039,7 @@ public void handle(LeaveGroupResponse leaveResponse, RequestFuture future) log.debug("LeaveGroup response with {} returned successfully: {}", sentGeneration, response); future.complete(null); } else { - log.error("LeaveGroup response with {} failed with error: {}", sentGeneration, error.message()); + log.error("LeaveGroup request with {} failed with error: {}", sentGeneration, error.message()); future.raise(error); } } @@ -1085,14 +1085,14 @@ public void handle(HeartbeatResponse heartbeatResponse, RequestFuture futu error == Errors.UNKNOWN_MEMBER_ID || error == Errors.FENCED_INSTANCE_ID) { if (generationUnchanged()) { - log.info("Attempt to heartbeat failed with generation {} and group instance id {} due to {}, resetting generation", - sentGeneration, rebalanceConfig.groupInstanceId, error.message()); + log.info("Attempt to heartbeat with {} and group instance id {} failed due to {}, resetting generation", + sentGeneration, rebalanceConfig.groupInstanceId, error); resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); future.raise(error); } else { // if the generation has changed, then ignore this error - log.info("Attempt to heartbeat failed with stale generation {} and group instance id {} due to {}, ignoring the error", - sentGeneration, rebalanceConfig.groupInstanceId, error.message()); + log.info("Attempt to heartbeat with stale {} and group instance id {} failed due to {}, ignoring the error", + sentGeneration, rebalanceConfig.groupInstanceId, error); future.complete(null); } } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) { From fc4c6888df3cc6dcfe6e88bb35bd9bb13f0cda0c Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Thu, 23 Apr 2020 11:32:16 -0700 Subject: [PATCH 9/9] github comments --- .../internals/AbstractCoordinatorTest.java | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index 3a27dd413a201..9f9c997eb29a8 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -486,7 +486,7 @@ public void testJoinGroupUnknownMemberResponseWithOldGeneration() throws Interru RequestFuture future = coordinator.sendJoinGroupRequest(); TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, - "The join-group request was not sent in time after"); + "The join-group request was not sent"); // change the generation after the join-group request final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -518,7 +518,7 @@ public void testSyncGroupUnknownMemberResponseWithOldGeneration() throws Interru consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); return !mockClient.requests().isEmpty(); }, 2000, - "The join-group request was not sent in time"); + "The join-group request was not sent"); mockClient.respond(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); assertTrue(mockClient.requests().isEmpty()); @@ -527,7 +527,7 @@ public void testSyncGroupUnknownMemberResponseWithOldGeneration() throws Interru consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); return !mockClient.requests().isEmpty(); }, 2000, - "The sync-group request was not sent in time"); + "The sync-group request was not sent"); // change the generation after the sync-group request final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -558,7 +558,7 @@ public void testSyncGroupIllegalGenerationResponseWithOldGeneration() throws Int consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); return !mockClient.requests().isEmpty(); }, 2000, - "The join-group request was not sent in time"); + "The join-group request was not sent"); mockClient.respond(joinGroupFollowerResponse(currGen.generationId, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); assertTrue(mockClient.requests().isEmpty()); @@ -567,7 +567,7 @@ public void testSyncGroupIllegalGenerationResponseWithOldGeneration() throws Int consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); return !mockClient.requests().isEmpty(); }, 2000, - "The sync-group request was not sent in time"); + "The sync-group request was not sent"); // change the generation after the sync-group request final AbstractCoordinator.Generation newGen = new AbstractCoordinator.Generation( @@ -595,7 +595,7 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int mockTime.sleep(HEARTBEAT_INTERVAL_MS); TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, - "The heartbeat request was not sent in time"); + "The heartbeat request was not sent"); assertTrue(coordinator.heartbeat().hasInflight()); // change the generation @@ -612,7 +612,7 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int coordinator.pollHeartbeat(mockTime.milliseconds()); return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time"); + "The heartbeat response was not received"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); @@ -629,7 +629,7 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru mockTime.sleep(HEARTBEAT_INTERVAL_MS); TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, - "The heartbeat request was not sent in time"); + "The heartbeat request was not sent"); assertTrue(coordinator.heartbeat().hasInflight()); // change the generation @@ -646,7 +646,7 @@ public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws Interru coordinator.pollHeartbeat(mockTime.milliseconds()); return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time"); + "The heartbeat response was not received"); // the generation should not be reset assertEquals(newGen, coordinator.generation()); @@ -663,7 +663,7 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru mockTime.sleep(HEARTBEAT_INTERVAL_MS); TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, - "The heartbeat request was not sent in time"); + "The heartbeat request was not sent"); assertTrue(coordinator.heartbeat().hasInflight()); @@ -677,7 +677,7 @@ public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws Interru return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time"); + "The heartbeat response was not received"); // the generation should be reset but the rebalance should still proceed assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation()); @@ -700,7 +700,7 @@ public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws Interr mockTime.sleep(HEARTBEAT_INTERVAL_MS); TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, - "The heartbeat request was not sent in time"); + "The heartbeat request was not sent"); assertTrue(coordinator.heartbeat().hasInflight()); // change the generation @@ -717,7 +717,7 @@ public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws Interr coordinator.pollHeartbeat(mockTime.milliseconds()); return !coordinator.heartbeat().hasInflight(); }, 2000, - "The heartbeat response was not received in time"); + "The heartbeat response was not received"); // the generation should not be reset assertEquals(newGen, coordinator.generation());