private void gracefulCloseTest(ConsumerCoordinator coordinator, boolean shouldLeaveGroup) {
        final AtomicBoolean commitRequested = new AtomicBoolean();
        final AtomicBoolean leaveGroupRequested = new AtomicBoolean();
        client.prepareResponse(body -> {
            commitRequested.set(true);
            OffsetCommitRequest commitRequest = (OffsetCommitRequest) body;
            return commitRequest.data().groupId().equals(groupId);
        }, new OffsetCommitResponse(new OffsetCommitResponseData()));
        if (shouldLeaveGroup)
            client.prepareResponse(body -> {
                leaveGroupRequested.set(true);
                LeaveGroupRequest leaveRequest = (LeaveGroupRequest) body;
                return leaveRequest.data().groupId().equals(groupId);
            }, new LeaveGroupResponse(new LeaveGroupResponseData()
                    .setErrorCode(Errors.NONE.code())));
        client.prepareResponse(body -> {
            commitRequested.set(true);
            OffsetCommitRequest commitRequest = (OffsetCommitRequest) body;
            return commitRequest.data().groupId().equals(groupId);
        }, new OffsetCommitResponse(new OffsetCommitResponseData()));

        coordinator.close();
        assertTrue(commitRequested.get(), "Commit not requested");
        assertEquals(shouldLeaveGroup, leaveGroupRequested.get(), "leaveGroupRequested should be " + shouldLeaveGroup);

        if (shouldLeaveGroup) {
            assertEquals(1, rebalanceListener.revokedCount);
            assertEquals(singleton(t1p), rebalanceListener.revoked);
        }
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java