@Test
public void testFlumeNodeReconnect() throws TTransportException, IOException,
    InterruptedException {

  FlumeConfiguration cfg = FlumeConfiguration.createTestableConfiguration();
  Clock.resetDefault();
  // Set directory of webapps to build-specific dir
  cfg.set(FlumeConfiguration.WEBAPPS_PATH, "build/webapps");
  // Doesn't matter whether or not we use ZK - use memory for speed
  cfg.set(FlumeConfiguration.MASTER_STORE, "memory");
  FlumeNode node = new FlumeNode(cfg);
  node.start();
  // avoiding gossip ack manager until it shuts down cleanly.
  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");
}