@Test
public void testRandomBackoff() throws Exception {

  Properties p = new Properties();
  List<LoadBalancedAvroHandler> hosts =
      new ArrayList<LoadBalancedAvroHandler>();
  List<Server> servers = new ArrayList<Server>();
  StringBuilder hostList = new StringBuilder("");
  for (int i = 0; i < 3; i++) {
    LoadBalancedAvroHandler s = new LoadBalancedAvroHandler();
    hosts.add(s);
    Server srv = RpcTestUtils.startServer(s);
    servers.add(srv);
    String name = "h" + i;
    p.put("hosts." + name, "127.0.0.1:" + srv.getPort());
    hostList.append(name).append(" ");
  }
  p.put("hosts", hostList.toString().trim());
  p.put("client.type", "default_loadbalance");
  p.put("host-selector", "random");
  p.put("backoff", "true");
  hosts.get(0).setFailed();
  hosts.get(2).setFailed();
  RpcClient c = RpcClientFactory.getInstance(p);
  Assert.assertTrue(c instanceof LoadBalancingRpcClient);
  // TODO: there is a remote possibility that s0 or s2
  // never get hit by the random assignment
  // and thus not backoffed, causing the test to fail
  for (int i = 0; i < 50; i++) {
    // a well behaved runner would always check the return.
    c.append(EventBuilder.withBody(("test" + String.valueOf(i)).getBytes()));
  }
  Assert.assertEquals(50, hosts.get(1).getAppendCount());
  Assert.assertEquals(0, hosts.get(0).getAppendCount());
  Assert.assertEquals(0, hosts.get(2).getAppendCount());
  hosts.get(0).setOK();
  hosts.get(1).setFailed(); // s0 should still be backed off
  try {
    c.append(EventBuilder.withBody("shouldfail".getBytes()));
    // nothing should be able to process right now
    Assert.fail("Expected EventDeliveryException");
  } catch (EventDeliveryException e) {
    // this is expected
  }
  Thread.sleep(2500); // wait for s0 to no longer be backed off
  for (int i = 0; i < 50; i++) {
    // a well behaved runner would always check the return.
    c.append(EventBuilder.withBody(("test" + String.valueOf(i)).getBytes()));
  }
  Assert.assertEquals(50, hosts.get(0).getAppendCount());
  Assert.assertEquals(50, hosts.get(1).getAppendCount());
  Assert.assertEquals(0, hosts.get(2).getAppendCount());
}