### Question:
FlumeNode implements Reportable { synchronized public void start() { FlumeConfiguration conf = FlumeConfiguration.get(); ReportManager.get().add(vmInfo); ReportManager.get().add(sysInfo); ReportManager.get().add(this); if (startHttp) { try { String webPath = getWebPath(conf); boolean findport = FlumeConfiguration.get().getNodeAutofindHttpPort(); this.http = new StatusHttpServer("flumeagent", webPath, "0.0.0.0", conf.getNodeStatusPort(), findport); http.start(); } catch (IOException e) { LOG.error("Flume node failed: " + e.getMessage(), e); } catch (Throwable t) { LOG.error("Unexcepted exception/error thrown! " + t.getMessage(), t); } } if (reportPusher != null) { reportPusher.start(); } if (liveMan != null) { liveMan.start(); } if (chokeMan != null) { chokeMan.setDaemon(true); chokeMan.start(); } } 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(); static void loadOutputFormatPlugins(); void init(String[] args); void destroy(); 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(); @Override ReportEvent getReport(); String getPhysicalNodeName();  }### Answer:
@Test public void testFlumeNodeReconnect() throws TTransportException, IOException, InterruptedException { FlumeConfiguration cfg = FlumeConfiguration.createTestableConfiguration(); Clock.resetDefault(); cfg.set(FlumeConfiguration.WEBAPPS_PATH, "build/webapps"); cfg.set(FlumeConfiguration.MASTER_STORE, "memory"); FlumeNode node = new FlumeNode(cfg); node.start(); ConfigStore cfgStore = FlumeMaster.createConfigStore(cfg); FlumeMaster fm = new FlumeMaster(new CommandManager(), new ConfigManager( cfgStore), new StatusManager(), new MasterAckManager(), cfg); assertEquals(0, fm.getKnownNodes().size()); fm.serve(); LOG.info("flume master 1 open "); while (fm.getKnownNodes().size() == 0) { Clock.sleep(1000); } fm.shutdown(); LOG.info("flume master 1 closed"); ConfigStore cfgStore2 = FlumeMaster.createConfigStore(cfg); FlumeMaster fm2 = new FlumeMaster(new CommandManager(), new ConfigManager( cfgStore2), new StatusManager(), new MasterAckManager(), cfg); assertEquals(0, fm2.getKnownNodes().size()); fm2.serve(); LOG.info("flume master 2 open "); while (fm2.getKnownNodes().size() == 0) { Clock.sleep(1000); } fm2.shutdown(); LOG.info("flume master 2 closed"); }