@Test
public void testCommitOffsetAsyncDisconnected() {

    client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
    coordinator.ensureCoordinatorReady();
    // async commit with coordinator disconnected
    MockCommitCallback cb = new MockCommitCallback();
    client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code())), true);
    coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), cb);
    assertTrue(coordinator.coordinatorUnknown());
    assertEquals(1, cb.invoked);
    assertTrue(cb.exception instanceof RetriableCommitFailedException);
}