@Test
public void testRetries() throws Exception {

    // create a sender with retries = 1
    int maxRetries = 1;
    Metrics m = new Metrics();
    try {
        Sender sender = new Sender(client,
                                   metadata,
                                   this.accumulator,
                                   false,
                                   MAX_REQUEST_SIZE,
                                   ACKS_ALL,
                                   maxRetries,
                                   m,
                                   time,
                                   "clientId",
                                   REQUEST_TIMEOUT);
        // do a successful retry
        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
        String id = client.requests().peek().request().destination();
        Node node = new Node(Integer.valueOf(id), "localhost", 0);
        assertEquals(1, client.inFlightRequestCount());
        assertTrue("Client ready status should be true", client.isReady(node, 0L));
        client.disconnect(id);
        assertEquals(0, client.inFlightRequestCount());
        assertFalse("Client ready status should be false", client.isReady(node, 0L));
        sender.run(time.milliseconds()); // receive error
        sender.run(time.milliseconds()); // reconnect
        sender.run(time.milliseconds()); // resend
        assertEquals(1, client.inFlightRequestCount());
        long offset = 0;
        client.respond(produceResponse(tp, offset, Errors.NONE.code(), 0));
        sender.run(time.milliseconds());
        assertTrue("Request should have retried and completed", future.isDone());
        assertEquals(offset, future.get().offset());
        // do an unsuccessful retry
        future = accumulator.append(tp, 0L, "key".getBytes(), "value".getBytes(), null, MAX_BLOCK_TIMEOUT).future;
        sender.run(time.milliseconds()); // send produce request
        for (int i = 0; i < maxRetries + 1; i++) {
            client.disconnect(client.requests().peek().request().destination());
            sender.run(time.milliseconds()); // receive error
            sender.run(time.milliseconds()); // reconnect
            sender.run(time.milliseconds()); // resend
        }
        sender.run(time.milliseconds());
        completedWithError(future, Errors.NETWORK_EXCEPTION);
    } finally {
        m.close();
    }
}