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 66da319645ca2..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 @@ -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 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 || 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 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.debug("SyncGroup failed: {}", error.message()); - resetGenerationOnResponseError(ApiKeys.SYNC_GROUP, error); + log.info("SyncGroup with {} failed: {}, 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 { @@ -890,8 +909,8 @@ protected synchronized String memberId() { } private synchronized void resetGeneration() { - this.rejoinNeeded = true; - this.generation = Generation.NO_GENERATION; + rejoinNeeded = true; + generation = Generation.NO_GENERATION; } synchronized void resetGenerationOnResponseError(ApiKeys api, Errors error) { @@ -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 request 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 @@ -1060,17 +1081,20 @@ 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) { - 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.", sentGeneration.memberId); - resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error); - future.raise(error); + } else if (error == Errors.ILLEGAL_GENERATION || + error == Errors.UNKNOWN_MEMBER_ID || + error == Errors.FENCED_INSTANCE_ID) { + if (generationUnchanged()) { + 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 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) { future.raise(GroupAuthorizationException.forGroupId(rebalanceConfig.groupId)); } else { @@ -1080,7 +1104,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 +1135,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,11 +1455,11 @@ private static class UnjoinedGroupException extends RetriableException { } // For testing only below - public Heartbeat heartbeat() { + final Heartbeat heartbeat() { return heartbeat; } - final void setLastRebalanceTime(final long timestamp) { + final synchronized void setLastRebalanceTime(final long timestamp) { lastRebalanceEndMs = timestamp; } @@ -1449,4 +1483,12 @@ final boolean hasUnknownGeneration() { final boolean hasValidMemberId() { return generation != Generation.NO_GENERATION && generation.hasMemberId(); } + + final synchronized void setNewGeneration(final Generation generation) { + this.generation = generation; + } + + final synchronized 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 b0f476a6157a5..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 @@ -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) @@ -1064,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(); @@ -1135,14 +1145,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; } @@ -1190,8 +1200,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, @@ -1209,9 +1227,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())); @@ -1253,6 +1280,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 b1a5a8c090567..8073ab4466aca 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 aa46c4b8a3d0c..7f02d2d25bc7e 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 @@ -60,6 +60,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..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 @@ -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(); @@ -466,24 +477,193 @@ public void testSyncGroupRequestWithFencedInstanceIdException() { } @Test - public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws InterruptedException { + public void testJoinGroupUnknownMemberResponseWithOldGeneration() throws InterruptedException { setupCoordinator(); - mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + joinGroup(); - final int generation = 1; + final AbstractCoordinator.Generation currGen = coordinator.generation(); - mockClient.prepareResponse(joinGroupFollowerResponse(generation, memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.NONE)); - mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + RequestFuture future = coordinator.sendJoinGroupRequest(); - coordinator.ensureActiveGroup(); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The join-group request was not sent"); + + // 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.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())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testSyncGroupUnknownMemberResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + joinGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); + RequestFuture future = coordinator.sendJoinGroupRequest(); + + TestUtils.waitForCondition(() -> { + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + return !mockClient.requests().isEmpty(); + }, 2000, + "The join-group request was not sent"); + + 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"); + + // 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.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())); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testSyncGroupIllegalGenerationResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + joinGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + coordinator.setNewState(AbstractCoordinator.MemberState.REBALANCING); + RequestFuture future = coordinator.sendJoinGroupRequest(); + + TestUtils.waitForCondition(() -> { + consumerClient.poll(mockTime.timer(REQUEST_TIMEOUT_MS)); + return !mockClient.requests().isEmpty(); + }, 2000, + "The join-group request was not sent"); + + 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"); + + // 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.respond(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(); + joinGroup(); 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"); + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent"); + 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 + TestUtils.waitForCondition(() -> { + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, + "The heartbeat response was not received"); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testHeartbeatUnknownMemberResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + joinGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent"); + 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 + TestUtils.waitForCondition(() -> { + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !coordinator.heartbeat().hasInflight(); + }, 2000, + "The heartbeat response was not received"); + + // the generation should not be reset + assertEquals(newGen, coordinator.generation()); + } + + @Test + public void testHeartbeatUnknownMemberResponseDuringRebalancing() throws InterruptedException { + setupCoordinator(); + joinGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent"); assertTrue(coordinator.heartbeat().hasInflight()); @@ -497,20 +677,52 @@ 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"); // 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(); assertEquals(currGen, coordinator.generation()); } + @Test + public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws InterruptedException { + setupCoordinator(); + joinGroup(); + + final AbstractCoordinator.Generation currGen = coordinator.generation(); + + // let the heartbeat request to send out a request + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000, + "The heartbeat request was not sent"); + 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.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"); + + // 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 diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala index 049dbef10addf..eacb432747a29 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala @@ -389,10 +389,10 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState 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)} during $operation, will " + - s"respond with instance fenced error") - true + 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 } 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 ed5ea0d108625..be080da3fe77f 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala @@ -639,10 +639,12 @@ class GroupCoordinatorTest { EasyMock.reset(replicaManager) val oldFollowerSyncGroupFuture = sendSyncGroupFollower(groupId, oldFollowerJoinGroupResult.generationId, oldFollowerJoinGroupResult.memberId, Some(protocolType), Some(protocolName), followerInstanceId) + EasyMock.reset(replicaManager) val duplicateFollowerJoinFuture = sendJoinGroup(groupId, JoinGroupRequest.UNKNOWN_MEMBER_ID, protocolType, protocols, groupInstanceId = followerInstanceId) timer.advanceClock(1) + val oldFollowerSyncGroupResult = Await.result(oldFollowerSyncGroupFuture, Duration(1, TimeUnit.MILLISECONDS)) assertEquals(Errors.FENCED_INSTANCE_ID, oldFollowerSyncGroupResult.error) assertTrue(getGroup(groupId).is(PreparingRebalance))