@Test
public void testMetadataChangeTriggersRebalance() { 

    final String consumerId = "consumer";
    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.needReassignment();
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
    coordinator.ensurePartitionAssignment();
    assertFalse(subscriptions.partitionAssignmentNeeded());
    // a new partition is added to the topic
    metadata.update(TestUtils.singletonCluster(topicName, 2), time.milliseconds());
    // we should detect the change and ask for reassignment
    assertTrue(subscriptions.partitionAssignmentNeeded());
}