@Test
public void testUnknownConsumerId() {

    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // illegal_generation will cause re-partition
    subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
    subscriptions.assignFromSubscribed(Collections.singletonList(tp));
    time.sleep(sessionTimeoutMs);
    RequestFuture<Void> future = coordinator.sendHeartbeatRequest(); // should send out the heartbeat
    assertEquals(1, consumerClient.pendingRequestCount());
    assertFalse(future.isDone());
    client.prepareResponse(heartbeatResponse(Errors.UNKNOWN_MEMBER_ID.code()));
    time.sleep(sessionTimeoutMs);
    consumerClient.poll(0);
    assertTrue(future.isDone());
    assertTrue(future.failed());
    assertEquals(Errors.UNKNOWN_MEMBER_ID.exception(), future.exception());
    assertTrue(coordinator.needRejoin());
}