@Test
public void testFailover() throws InterruptedException {

  Channel ch = new MemoryChannel();
  ConsumeXSink s1 = new ConsumeXSink(10);
  s1.setChannel(ch);
  s1.setName("s1");
  ConsumeXSink s2 = new ConsumeXSink(50);
  s2.setChannel(ch);
  s2.setName("s2");
  ConsumeXSink s3 = new ConsumeXSink(100);
  s3.setChannel(ch);
  s3.setName("s3");
  Context context = new Context();
  Configurables.configure(s1, context);
  Configurables.configure(s2, context);
  Configurables.configure(s3, context);
  Configurables.configure(ch, context);
  ch.start();
  List<Sink> sinks = new LinkedList<Sink>();
  sinks.add(s1);
  sinks.add(s2);
  sinks.add(s3);
  SinkGroup group = new SinkGroup(sinks);
  Map<String, String> params = new HashMap<String, String>();
  params.put("sinks", "s1 s2 s3");
  params.put("processor.type", "failover");
  params.put("processor.priority.s1", "3");
  params.put("processor.priority.s2", "2");
  params.put("processor.priority.s3", "1");
  params.put("processor.maxpenalty", "10000");
  context.putAll(params);
  Configurables.configure(group, context);
  SinkRunner runner = new SinkRunner(group.getProcessor());
  runner.start();
  Assert.assertEquals(LifecycleState.START, s1.getLifecycleState());
  Assert.assertEquals(LifecycleState.START, s2.getLifecycleState());
  Assert.assertEquals(LifecycleState.START, s3.getLifecycleState());
  for (int i = 0; i < 15; i++) {
    Transaction tx = ch.getTransaction();
    tx.begin();
    ch.put(EventBuilder.withBody("test".getBytes()));
    tx.commit();
    tx.close();
  }
  Thread.sleep(100);
  Assert.assertEquals(new Integer(10), s1.getWritten());
  Assert.assertEquals(new Integer(5), s2.getWritten());
  for (int i = 0; i < 50; i++) {
    Transaction tx = ch.getTransaction();
    tx.begin();
    ch.put(EventBuilder.withBody("test".getBytes()));
    tx.commit();
    tx.close();
  }
  Thread.sleep(100);
  Assert.assertEquals(new Integer(50), s2.getWritten());
  Assert.assertEquals(new Integer(5), s3.getWritten());
  // test rollover to recovered servers
  s2.setRemaining(20);
  // get us past the retry time for the failed sink
  Thread.sleep(5000);
  for (int i = 0; i < 100; i++) {
    Transaction tx = ch.getTransaction();
    tx.begin();
    ch.put(EventBuilder.withBody("test".getBytes()));
    tx.commit();
    tx.close();
  }
  Thread.sleep(1000);
  Assert.assertEquals(new Integer(10), s1.getWritten());
  Assert.assertEquals(new Integer(70), s2.getWritten());
  Assert.assertEquals(new Integer(85), s3.getWritten());
  runner.stop();
  ch.stop();
}