@Test
public void testIllegalGenerationOnSyncGroup() {

    final String consumerId = "consumer";
    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.needReassignment();
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // join initially, but let coordinator rebalance on sync
    client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(Collections.<TopicPartition>emptyList(), Errors.ILLEGAL_GENERATION.code()));
    // then let the full join/sync finish successfully
    client.prepareResponse(new MockClient.RequestMatcher() {
        @Override
        public boolean matches(ClientRequest request) {
            JoinGroupRequest joinRequest = new JoinGroupRequest(request.request().body());
            return joinRequest.memberId().equals(JoinGroupRequest.UNKNOWN_MEMBER_ID);
        }
    }, joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
    coordinator.ensurePartitionAssignment();
    assertFalse(subscriptions.partitionAssignmentNeeded());
    assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions());
}