@Test
public void testRandomBackoff() throws Exception {

  Channel ch = new MockChannel();
  int n = 100;
  int numEvents = n;
  for (int i = 0; i < numEvents; i++) {
    ch.put(new MockEvent("test" + i));
  }
  MockSink s1 = new MockSink(1);
  s1.setChannel(ch);
  // s1 always fails
  s1.setFail(true);
  MockSink s2 = new MockSink(2);
  s2.setChannel(ch);
  MockSink s3 = new MockSink(3);
  s3.setChannel(ch);
  // s3 always fails
  s3.setFail(true);
  List<Sink> sinks = new ArrayList<Sink>();
  sinks.add(s1);
  sinks.add(s2);
  sinks.add(s3);
  LoadBalancingSinkProcessor lbsp = getProcessor("random", sinks, true);
  // 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.
    lbsp.process();
  }
  Assert.assertEquals(50, s2.getEvents().size());
  s2.setFail(true);
  s1.setFail(false); // s1 should still be backed off
  try {
    lbsp.process();
    // nothing should be able to process right now
    Assert.fail("Expected EventDeliveryException");
  } catch (EventDeliveryException e) {
    // this is expected
  }
  Thread.sleep(2100); // wait for s1 to no longer be backed off
  Sink.Status s = Sink.Status.READY;
  while (s != Sink.Status.BACKOFF) {
    s = lbsp.process();
  }
  Assert.assertEquals(50, s1.getEvents().size());
  Assert.assertEquals(50, s2.getEvents().size());
  Assert.assertEquals(0, s3.getEvents().size());
}