@Test
public void testSimple() throws Exception {

    long offset = 0;
    Future<RecordMetadata> future = accumulator.append(tp, 0L, "key".getBytes(), "value".getBytes(), null, MAX_BLOCK_TIMEOUT).future;
    sender.run(time.milliseconds()); // connect
    sender.run(time.milliseconds()); // send produce request
    assertEquals("We should have a single produce request in flight.", 1, client.inFlightRequestCount());
    client.respond(produceResponse(tp, offset, Errors.NONE.code(), 0));
    sender.run(time.milliseconds());
    assertEquals("All requests completed.", offset, (long) client.inFlightRequestCount());
    sender.run(time.milliseconds());
    assertTrue("Request should be completed", future.isDone());
    assertEquals(offset, future.get().offset());
}