  @Test
  public void testRandomPersistentFailure() throws Exception {

    Channel ch = new MockChannel();
    int n = 100;
    int numEvents = 3*n;
    for (int i = 0; i < numEvents; i++) {
      ch.put(new MockEvent("test" + i));
    }

    MockSink s1 = new MockSink(1);
    s1.setChannel(ch);

    MockSink s2 = new MockSink(2);
    s2.setChannel(ch);

    // s2 always fails
    s2.setFail();

    MockSink s3 = new MockSink(3);
    s3.setChannel(ch);

    List<Sink> sinks = new ArrayList<Sink>();
    sinks.add(s1);
    sinks.add(s2);
    sinks.add(s3);

    LoadBalancingSinkProcessor lbsp = getProcessor("random",sinks);

    Status s = Status.READY;
    while (s != Status.BACKOFF) {
      s = lbsp.process();
    }

    Assert.assertTrue(s2.getEvents().size() == 0);
    Assert.assertTrue(s1.getEvents().size() + s3.getEvents().size() == 3*n);
  }