@Test
public void testRestartNextException() throws Exception {

  LogicalNode node = new LogicalNode(new Context(), "test");
  final AtomicInteger count = new AtomicInteger();
  class NextExnSource extends EventSource.Base {
    @Override
    public void close() throws IOException {
      System.out.println("Next Exn Source closed");
    }
    @Override
    public Event next() throws IOException {
      int i = count.incrementAndGet();
      if (i > 100) {
        return null; // close naturally
      }
      // "fail"
      throw new IOException("ayiii!");
    }
    @Override
    public void open() throws IOException {
      System.out.println("Next Exn Source opened");
    }
  }
  ;
  FlumeBuilder.setSourceFactory(new SourceFactory() {
    public EventSource getSource(Context ctx, String name, String... args) {
      return new NextExnSource();
    }
    @Override
    public Set<String> getSourceNames() {
      HashSet<String> set = new HashSet<String>();
      set.add("newExnSource");
      return set;
    }
  });
  // 0 is the first cfg version
  FlumeConfigData cfg = new FlumeConfigData(0, "null", "null", 0, 0,
      "my-test-flow");
  node.loadConfig(cfg); // this will load the NextExnSource and a NullSink
  ReportEvent rpt = node.getMetrics();
  LOG.info(rpt.toString());
  // This node should exit in error state since next eventually will only
  // error out with exceptions.
  assertTrue(node.getDriver().waitForState(DriverState.ERROR, 1000));
  System.out.printf("next called %d times", count.get());
  System.out.flush();
  assertEquals(1, count.get());
}