private void prepareJoinAndSyncResponse(String consumerId, int generation, List<String> subscription, List<TopicPartition> assignment) {
        partitionAssignor.prepare(singletonMap(consumerId, assignment));
        client.prepareResponse(
                joinGroupLeaderResponse(
                        generation, consumerId, singletonMap(consumerId, subscription), Errors.NONE));
        client.prepareResponse(body -> {
            SyncGroupRequest sync = (SyncGroupRequest) body;
            return sync.data().memberId().equals(consumerId) &&
                    sync.data().generationId() == generation &&
                    sync.groupAssignments().containsKey(consumerId);
        }, syncGroupResponse(assignment, Errors.NONE));
    }
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