FlumeNode implements Reportable { @Override public ReportEvent getMetrics() { ReportEvent node = new ReportEvent(getName()); node.setLongMetric(R_NUM_LOGICAL_NODES, this.getLogicalNodeManager() .getNodes().size()); return node; } FlumeNode(String name, MasterRPC rpc, LogicalNodeManager nodesMan,
      WALManager walMan, DiskFailoverManager dfMan,
      CollectorAckListener colAck, LivenessManager liveman);  FlumeNode(FlumeConfiguration conf, String nodeName, MasterRPC rpc,
      boolean startHttp, boolean oneshot);  FlumeNode(MasterRPC rpc, boolean startHttp, boolean oneshot);  FlumeNode(String nodename, FlumeConfiguration conf, boolean startHttp,
      boolean oneshot);  FlumeNode(FlumeConfiguration conf); static FlumeNode getInstance(); static String getWebPath(FlumeConfiguration conf); synchronized void start(); synchronized void stop(); void init(String[] args); void destroy(); @Deprecated String report(); WALAckManager getAckChecker(); AckListener getCollectorAckListener(); static void logVersion(Logger log); static void logEnvironment(Logger log); static void nodeConfigChecksOk(); static FlumeNode setup(String[] argv); static void main(String[] argv); WALManager getWalManager(); WALManager getWalManager(String walnode); WALManager addWalManager(String walnode); WALManager getAddWALManager(String dfonode); DiskFailoverManager getDFOManager(); DiskFailoverManager getDFOManager(String dfonode); DiskFailoverManager addDFOManager(String dfonode); DiskFailoverManager getAddDFOManager(String dfonode); LogicalNodeManager getLogicalNodeManager(); LivenessManager getLivenessManager(); ChokeManager getChokeManager(); @Override String getName(); @Deprecated ReportEvent getReport(); @Override ReportEvent getMetrics(); @Override Map<String, Reportable> getSubMetrics(); String getPhysicalNodeName(); SystemInfo getSystemInfo(); FlumeVMInfo getVMInfo();  }
@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; } 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; } }); FlumeConfigData cfg = new FlumeConfigData(0, "null", "null", 0, 0, "my-test-flow"); node.loadConfig(cfg); ReportEvent rpt = node.getMetrics(); LOG.info(rpt.toString()); assertTrue(node.getDriver().waitForState(DriverState.ERROR, 1000)); System.out.printf("next called %d times", count.get()); System.out.flush(); assertEquals(1, count.get()); }