@Test
public void testAutoCommitDynamicAssignmentRebalance() {

    final String consumerId = "consumer";
    ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors,
            ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true);
    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.needReassignment();
    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // haven't joined, so should not cause a commit
    time.sleep(autoCommitIntervalMs);
    consumerClient.poll(0);
    client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code()));
    client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
    coordinator.ensurePartitionAssignment();
    subscriptions.seek(tp, 100);
    client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code())));
    time.sleep(autoCommitIntervalMs);
    consumerClient.poll(0);
    assertEquals(100L, subscriptions.committed(tp).offset());
}