@Test
public void testDisconnectInJoin() {

    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.needReassignment();
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // disconnected from original coordinator will cause re-discover and join again
    client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()), true);
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
    coordinator.ensurePartitionAssignment();
    assertFalse(subscriptions.partitionAssignmentNeeded());
    assertEquals(Collections.singleton(tp), subscriptions.assignedPartitions());
    assertEquals(1, rebalanceListener.revokedCount);
    assertEquals(1, rebalanceListener.assignedCount);
    assertEquals(Collections.singleton(tp), rebalanceListener.assigned);
}