public class FlumeNode implements Reportable {
  static final Logger LOG = LoggerFactory.getLogger(FlumeNode.class);
  final static String PHYSICAL_NODE_REPORT_PREFIX = "pn-";
  static final String R_NUM_LOGICAL_NODES = "Logical nodes";

  // hook for jsp/web display
  private static FlumeNode instance;

  // run mode for this node.
  final boolean startHttp; // should this node start its own http status server
  private InternalHttpServer http = null;

  private FlumeVMInfo vmInfo;
  private SystemInfo sysInfo;
  private final LivenessManager liveMan;
  private MasterRPC rpcMan;
  private LogicalNodeManager nodesMan;

  private ReportManager simpleReportManager = new ReportManager("simple");
  private final MasterReportPusher reportPusher;

  /**
   * (String) logical node name -> WALManager
   */
  private Map<String, WALManager> walMans = new HashMap<String, WALManager>();

  /**
   * (String) logical node name -> DiskFailoverManager
   */
  private Map<String, DiskFailoverManager> failoverMans = new HashMap<String, DiskFailoverManager>();

  final private CollectorAckListener collectorAck;

  final String physicalNodeName;

  private final ChokeManager chokeMan;

  /**
   * A FlumeNode constructor with pluggable xxxManagers. This is used for
   * debugging and test cases. The http server is assumed not to be started, and
   * we are not doing a oneshot.
   */
  public FlumeNode(String name, MasterRPC rpc, LogicalNodeManager nodesMan,
      WALManager walMan, DiskFailoverManager dfMan,
      CollectorAckListener colAck, LivenessManager liveman) {
    this.physicalNodeName = name;
    rpcMan = rpc;
    instance = this;
    this.startHttp = false;
    this.nodesMan = nodesMan;
    this.walMans.put(getPhysicalNodeName(), walMan);
    this.failoverMans.put(getPhysicalNodeName(), dfMan);
    this.collectorAck = colAck;
    this.liveMan = liveman;
    // As this is only for the testing puposes, just initialize the physical
    // node limit to Max Int.
    this.chokeMan = new ChokeManager();
    this.vmInfo = new FlumeVMInfo(PHYSICAL_NODE_REPORT_PREFIX
        + this.physicalNodeName + ".");

    this.reportPusher = new MasterReportPusher(FlumeConfiguration.get(),
        simpleReportManager, rpcMan);
    this.sysInfo = new SystemInfo(PHYSICAL_NODE_REPORT_PREFIX
        + this.physicalNodeName + ".");
  }

  public FlumeNode(FlumeConfiguration conf, String nodeName, MasterRPC rpc,
      boolean startHttp, boolean oneshot) {
    this.physicalNodeName = nodeName;
    rpcMan = rpc;
    instance = this;
    this.startHttp = startHttp;
    this.nodesMan = new LogicalNodeManager(nodeName);

    File defaultDir = new File(conf.getAgentLogsDir(), getPhysicalNodeName());
    WALManager walMan = new NaiveFileWALManager(defaultDir);
    this.walMans.put(getPhysicalNodeName(), walMan);
    this.failoverMans.put(getPhysicalNodeName(), new NaiveFileFailoverManager(
        defaultDir));

    // no need for liveness tracker if a one shot execution.
    this.collectorAck = new CollectorAckListener(rpcMan);
    if (!oneshot) {
      this.liveMan = new LivenessManager(nodesMan, rpcMan,
          new FlumeNodeWALNotifier(this.walMans));
      this.reportPusher = new MasterReportPusher(conf, simpleReportManager,
          rpcMan);

    } else {
      this.liveMan = null;
      this.reportPusher = null;
    }

    // initializing ChokeController
    this.chokeMan = new ChokeManager();

    this.vmInfo = new FlumeVMInfo(PHYSICAL_NODE_REPORT_PREFIX
        + this.getPhysicalNodeName() + ".");
    this.sysInfo = new SystemInfo(PHYSICAL_NODE_REPORT_PREFIX
        + this.getPhysicalNodeName() + ".");

  }

  public FlumeNode(MasterRPC rpc, boolean startHttp, boolean oneshot) {
    this(FlumeConfiguration.get(), NetUtils.localhost(), rpc, startHttp,
        oneshot);
  }

  public FlumeNode(String nodename, FlumeConfiguration conf, boolean startHttp,
      boolean oneshot) {
    // Use a failover-enabled master RPC, which randomizes the failover order
    this(conf, nodename, new MultiMasterRPC(conf, true), startHttp, oneshot);
  }

  public FlumeNode(FlumeConfiguration conf) {
    this(NetUtils.localhost(), conf, false /* http server */, false /* oneshot */);
  }

  /**
   * This hook makes it easy for web apps and jsps to get the current FlumeNode
   * instance. This is used to test the FlumeNode related jsps.
   */
  public static FlumeNode getInstance() {
    if (instance == null) {
      instance = new FlumeNode(FlumeConfiguration.get());
    }
    return instance;
  }

  public static String getWebPath(FlumeConfiguration conf) {
    String webPath = conf.getWebAppsPath();
    File f = new File(webPath);
    // absolute paths win, but if is not absolute, prefix with flume home
    if (!f.isAbsolute()) {
      String basepath = FlumeConfiguration.getFlumeHome();
      if (basepath == null) {
        LOG.warn("FLUME_HOME not set, potential for odd behavior!");
      }
      File base = new File(basepath, webPath);
      webPath = base.getAbsolutePath();
    }
    return webPath;
  }

  /**
   * This also implements the Apache Commons Daemon interface's start
   */
  synchronized public void start() {
    FlumeConfiguration conf = FlumeConfiguration.get();

    // the simple report interface
    simpleReportManager.add(vmInfo);
    simpleReportManager.add(sysInfo);
    simpleReportManager.add(new Reportable() {

      @Override
      public String getName() {
        return FlumeNode.this.getName();
      }

      @Override
      public ReportEvent getMetrics() {
        return FlumeNode.this.getReport();
      }

      @Override
      public Map<String, Reportable> getSubMetrics() {
        return ReportUtil.noChildren();
      }
    });

    // the full report interface
    ReportManager.get().add(vmInfo);
    ReportManager.get().add(sysInfo);
    ReportManager.get().add(this);

    if (startHttp) {
      try {
        http = new InternalHttpServer();

        http.addHandler(InternalHttpServer.createLogAppContext());

        http.addHandler(InternalHttpServer.createServletContext(
            StackServlet.class, "/stacks", "/*", "stacks"));

        http.setBindAddress("0.0.0.0");
        http.setPort(conf.getNodeStatusPort());
        String webAppRoot = FlumeConfiguration.get().getNodeWebappRoot();
        http.setWebappDir(new File(webAppRoot));
        http.setScanForApps(true);

        http.start();
      } catch (Throwable t) {
        LOG.error("Unexpected exception/error thrown! " + t.getMessage(), t);
      }
    }

    if (reportPusher != null) {
      reportPusher.start();
    }

    if (liveMan != null) {
      liveMan.start();
    }

    if (chokeMan != null) {
      // JVM exits if only daemons threads remain.
      chokeMan.setDaemon(true);
      chokeMan.start();
    }

  }

  /**
   * This also implements the Apache Commons Daemon interface's stop
   */
  synchronized public void stop() {
    if (this.http != null) {
      try {
        http.stop();
      } catch (Exception e) {
        LOG.error("Stopping http server failed: " + e);
      }
    }

    if (reportPusher != null) {
      reportPusher.stop();
    }

    if (liveMan != null) {
      liveMan.stop();
    }

    if (chokeMan != null) {
      chokeMan.halt();
    }

  }

  /**
   * This also implements the Apache Commons Daemon interface's init
   */
  public void init(String[] args) {
    try {
      setup(args);
    } catch (IOException ioe) {
      LOG.error("Failed to init Flume Node", ioe);
    }
  }

  /**
   * This also implements the Apache Commons Daemon interface's destroy
   */
  public void destroy() {
    stop(); // I think this is ok.
  }

  /**
   * This method is currently called by the JSP to display node information.
   */
  @Deprecated
  public String report() {
    return ReportUtil.getFlattenedReport(this).toHtml();
  }

  public WALAckManager getAckChecker() {
    if (liveMan == null)
      return null;
    return liveMan.getAckChecker();
  }

  public AckListener getCollectorAckListener() {
    return collectorAck;
  }

  public static void logVersion(Logger log) {
    log.info("Flume " + VersionInfo.getVersion());
    log.info(" rev " + VersionInfo.getRevision());
    log.info("Compiled  on " + VersionInfo.getDate());
  }

  public static void logEnvironment(Logger log) {
    Properties props = System.getProperties();
    for (Entry<Object, Object> p : props.entrySet()) {
      log.info("System property " + p.getKey() + "=" + p.getValue());
    }
  }

  /**
   * This function checks the agent logs dir to make sure that the process has
   * the ability to the directory if necessary, that the path if it does exist is
   * a directory, and that it can in fact create files inside of the directory.
   * If it fails any of these, it throws an exception.
   * 
   * Finally, it checks to see if the path is in /tmp and warns the user that
   * this may not be the best idea.
   */
  public static void nodeConfigChecksOk() throws IOException {
    // TODO (jon) if we add more checks in here, make the different managers
    // responsible for throwing an Exception on construction instead.

    FlumeConfiguration conf = FlumeConfiguration.get();

    String s = conf.getAgentLogsDir();
    File f = new File(s);

    if (!FileUtil.makeDirs(f)) {
      throw new IOException("Path to Log dir cannot be created: '" + s
          + "'.  Check permissions?");
    }

    if (!f.isDirectory()) {
      throw new IOException("Log dir '" + s
          + "' already exists as a file.  Check log dir path.");
    }

    File f2 = null;
    try {
      f2 = File.createTempFile("initcheck", ".test", f);
    } catch (IOException e) {
      throw new IOException("Failure to write in log directory: '" + s
          + "'.  Check permissions?");
    }
    if (!f2.delete()) {
      throw new IOException("Unable to delete " + f2 + " from log directory "
          + "(but writing succeeded) - something is strange here");
    }

    File tmp = new File("/tmp");
    File cur = f;
    while (cur != null) {
      if (cur.equals(tmp)) {
        LOG.warn("Log directory is writing inside of /tmp.  This data may not survive reboot!");
        break;
      }
      cur = cur.getParentFile();
    }
  }

  /**
   * Returns a Flume Node with settings from specified command line parameters.
   * (See usage for instructions)
   * 
   * @param argv
   * @return
   * @throws IOException
   */
  public static FlumeNode setup(String[] argv) throws IOException {
    logVersion(LOG);
    logEnvironment(LOG);
    // Make sure the Java version is not older than 1.6
    if (!CheckJavaVersion.isVersionOk()) {
      LOG.error("Exiting because of an old Java version or Java version in bad format");
      System.exit(-1);
    }
    LOG.info("Starting flume agent on: " + NetUtils.localhost());
    LOG.info(" Working directory is: " + new File(".").getAbsolutePath());

    FlumeConfiguration.hardExitLoadConfig(); // will exit if conf file is bad.

    CommandLine cmd = null;
    Options options = new Options();
    options.addOption("c", true, "Load initial config from cmdline arg");
    options.addOption("n", true, "Set node name");
    options.addOption("s", false,
        "Do not start local flume status server on node");
    options.addOption("1", false,
        "Make flume node one shot (if closes or errors, exits)");
    options.addOption("m", false,
        "Have flume hard exit if in likely GC thrash situation");
    options.addOption("h", false, "Print help information");
    options.addOption("v", false, "Print version information");
    try {
      CommandLineParser parser = new PosixParser();
      cmd = parser.parse(options, argv);
    } catch (ParseException e) {
      HelpFormatter fmt = new HelpFormatter();
      fmt.printHelp("FlumeNode", options, true);
      return null;
    }

    // dump version info only
    if (cmd != null && cmd.hasOption("v")) {
      return null;
    }

    // dump help info.
    if (cmd != null && cmd.hasOption("h")) {
      HelpFormatter fmt = new HelpFormatter();
      fmt.printHelp("FlumeNode", options, true);
      return null;
    }
    // Check FlumeConfiguration file for settings that may cause node to fail.
    nodeConfigChecksOk();

    String nodename = NetUtils.localhost(); // default to local host name.
    if (cmd != null && cmd.hasOption("n")) {
      // select a different name, allow for multiple processes configured
      // differently on same node.
      nodename = cmd.getOptionValue("n");
    }

    boolean startHttp = false;
    if (cmd != null && !cmd.hasOption("s")) {
      // no -s option, start the local status server
      startHttp = true;
    }

    boolean oneshot = false;
    if (cmd != null && cmd.hasOption("1")) {
      oneshot = true;
    }

    FormatFactory.loadOutputFormatPlugins();

    // Instantiate the flume node.
    FlumeConfiguration conf = FlumeConfiguration.get();

    FlumeNode flume = new FlumeNode(nodename, conf, startHttp, oneshot);

    flume.start();

    // load an initial configuration from command line
    if (cmd != null && cmd.hasOption("c")) {
      String spec = cmd.getOptionValue("c");
      LOG.info("Loading spec from command line: '" + spec + "'");

      try {
        // TODO the first one should be physical node name
        Context ctx = new LogicalNodeContext(nodename, nodename);
        Map<String, Pair<String, String>> cfgs = FlumeBuilder.parseConf(ctx,
            spec);
        Pair<String, String> node = cfgs.get(nodename);
        flume.nodesMan.spawn(nodename, node.getLeft(), node.getRight());
      } catch (Exception e) {
        LOG.warn("Caught exception loading node:" + e.getMessage());
        LOG.debug("Exception: ", e);
        if (oneshot) {
          System.exit(0); // exit cleanly
        }
      }

    } else {
      // default to null configurations.
      try {
        flume.nodesMan.spawn(nodename, "null", "null");
      } catch (FlumeSpecException e) {
        LOG.error("This should never happen", e);
      } catch (IOException e) {
        LOG.error("Caught exception loading node", e);
      }
    }

    if (cmd != null && cmd.hasOption("m")) {
      // setup memory use monitor
      LOG.info("Setup hard exit on memory exhaustion");
      MemoryMonitor.setupHardExitMemMonitor(FlumeConfiguration.get()
          .getAgentMemoryThreshold());
    }

    try {
      tryKerberosLogin();
    } catch (IOException ioe) {
      LOG.error("Failed to kerberos login.", ioe);
    }

    // hangout, waiting for other agent thread to exit.
    return flume;
  }

  /**
   * This should attempts to kerberos login via a keytab if security is enabled
   * in hadoop.
   * 
   * This should be able to support multiple hadoop clusters as long as the
   * particular principal is allowed on multiple clusters.
   * 
   * To preserve compatibility with non security enhanced hdfs, we use
   * reflection on various UserGroupInformation and SecurityUtil related method
   * calls.
   */
  @SuppressWarnings("unchecked")
  static void tryKerberosLogin() throws IOException {

    /*
     * UserGroupInformation is in hadoop 0.18
     * UserGroupInformation.isSecurityEnabled() not in pre security API.
     * 
     * boolean useSec = UserGroupInformation.isSecurityEnabled();
     */
    boolean useSec = false;

    try {
      Class<UserGroupInformation> c = UserGroupInformation.class;
      // static call, null this obj
      useSec = (Boolean) c.getMethod("isSecurityEnabled").invoke(null);
    } catch (Exception e) {
      LOG.warn("Flume is using Hadoop core "
          + org.apache.hadoop.util.VersionInfo.getVersion()
          + " which does not support Security / Authentication: "
          + e.getMessage());
      return;
    }

    LOG.info("Hadoop Security enabled: " + useSec);
    if (!useSec) {
      return;
    }

    // At this point we know we are using a hadoop library that is kerberos
    // enabled.

    // attempt to load kerberos information for authenticated hdfs comms.
    String principal = FlumeConfiguration.get().getKerberosPrincipal();
    String keytab = FlumeConfiguration.get().getKerberosKeytab();
    LOG.info("Kerberos login as " + principal + " from " + keytab);

    try {
      /*
       * SecurityUtil not present pre hadoop 20.2
       * 
       * SecurityUtil.login not in pre-security Hadoop API
       * 
       * // Keytab login does not need to auto refresh
       * 
       * SecurityUtil.login(FlumeConfiguration.get(),
       * FlumeConfiguration.SECURITY_KERBEROS_KEYTAB,
       * FlumeConfiguration.SECURITY_KERBEROS_PRINCIPAL);
       */
      Class c = Class.forName("org.apache.hadoop.security.SecurityUtil");
      // get method login(Configuration, String, String);
      Method m = c.getMethod("login", Configuration.class, String.class,
          String.class);
      m.invoke(null, FlumeConfiguration.get(),
          FlumeConfiguration.SECURITY_KERBEROS_KEYTAB,
          FlumeConfiguration.SECURITY_KERBEROS_PRINCIPAL);
    } catch (Exception e) {
      LOG.error("Flume failed when attempting to authenticate with keytab "
          + FlumeConfiguration.get().getKerberosKeytab() + " and principal '"
          + FlumeConfiguration.get().getKerberosPrincipal() + "'", e);

      // e.getMessage() comes from hadoop is worthless
      return;
    }

    try {
      /*
       * getLoginUser, getAuthenticationMethod, and isLoginKeytabBased are not
       * in Hadoop 20.2, only kerberized enhanced version.
       * 
       * getUserName is in all 0.18.3+
       * 
       * UserGroupInformation ugi = UserGroupInformation.getLoginUser();
       * LOG.info("Auth method: " + ugi.getAuthenticationMethod());
       * LOG.info(" User name: " + ugi.getUserName());
       * LOG.info(" Using keytab: " +
       * UserGroupInformation.isLoginKeytabBased());
       */

      Class<UserGroupInformation> c2 = UserGroupInformation.class;
      // static call, null this obj
      UserGroupInformation ugi = (UserGroupInformation) c2.getMethod(
          "getLoginUser").invoke(null);
      String authMethod = c2.getMethod("getAuthenticationMethod").invoke(ugi)
          .toString();
      boolean keytabBased = (Boolean) c2.getMethod("isLoginKeytabBased")
          .invoke(ugi);

      LOG.info("Auth method: " + authMethod);
      LOG.info(" User name: " + ugi.getUserName());
      LOG.info(" Using keytab: " + keytabBased);
    } catch (Exception e) {
      LOG.error("Flume was unable to dump kerberos login user"
          + " and authentication method", e);
      return;
    }

  }

  public static void main(String[] argv) {
    try {
      setup(argv);
    } catch (Exception e) {
      LOG.error(
          "Aborting: Unexpected problem with environment." + e.getMessage(), e);
      System.exit(-1);
    }
  }

  public WALManager getWalManager() {
    // default to physical node's wal manager
    synchronized (walMans) {
      return walMans.get(getPhysicalNodeName());
    }
  }

  public WALManager getWalManager(String walnode) {
    synchronized (walMans) {
      if (walnode == null) {
        return getWalManager();
      }
      return walMans.get(walnode);
    }
  }

  public WALManager addWalManager(String walnode) {
    Preconditions.checkArgument(walnode != null);
    FlumeConfiguration conf = FlumeConfiguration.get();
    WALManager wm = new NaiveFileWALManager(new File(new File(
        conf.getAgentLogsDir()), walnode));
    synchronized (walMans) {
      walMans.put(walnode, wm);
      return wm;
    }
  }

  /**
   * Atomically gets an existing dfo for the given node or creates a new one.
   */
  public WALManager getAddWALManager(String dfonode) {
    synchronized (failoverMans) {
      WALManager walman = getWalManager(dfonode);

      if (walman == null) {
        walman = addWalManager(dfonode);
      }
      return walman;
    }
  }

  public DiskFailoverManager getDFOManager() {
    synchronized (failoverMans) {
      return failoverMans.get(getPhysicalNodeName());
    }
  }

  public DiskFailoverManager getDFOManager(String dfonode) {
    synchronized (failoverMans) {
      if (dfonode == null) {
        return getDFOManager();
      }
      return failoverMans.get(dfonode);
    }
  }

  public DiskFailoverManager addDFOManager(String dfonode) {
    Preconditions.checkArgument(dfonode != null);
    FlumeConfiguration conf = FlumeConfiguration.get();
    DiskFailoverManager wm = new NaiveFileFailoverManager(new File(new File(
        conf.getAgentLogsDir()), dfonode));
    synchronized (failoverMans) {
      failoverMans.put(dfonode, wm);
      return wm;
    }
  }

  /**
   * Atomically gets an existing dfo for the given node or creates a new one.
   */
  public DiskFailoverManager getAddDFOManager(String dfonode) {
    synchronized (failoverMans) {
      DiskFailoverManager dfoman = getDFOManager(dfonode);

      if (dfoman == null) {
        dfoman = addDFOManager(dfonode);
      }
      return dfoman;
    }
  }

  public LogicalNodeManager getLogicalNodeManager() {
    return nodesMan;
  }

  // TODO (jon) rename when liveness manager renamed
  public LivenessManager getLivenessManager() {
    return liveMan;
  }

  public ChokeManager getChokeManager() {
    return chokeMan;
  }

  @Override
  public String getName() {
    return PHYSICAL_NODE_REPORT_PREFIX + this.getPhysicalNodeName();
  }

  @Deprecated
  public ReportEvent getReport() {
    ReportEvent node = new ReportEvent(getName());
    node.setLongMetric(R_NUM_LOGICAL_NODES, this.getLogicalNodeManager()
        .getNodes().size());
    node.hierarchicalMerge(nodesMan.getName(), nodesMan.getReport());
    if (getAckChecker() != null) {
      node.hierarchicalMerge(getAckChecker().getName(), getAckChecker()
          .getMetrics());
    }
    return node;
  }

  @Override
  public ReportEvent getMetrics() {
    ReportEvent node = new ReportEvent(getName());
    node.setLongMetric(R_NUM_LOGICAL_NODES, this.getLogicalNodeManager()
        .getNodes().size());
    return node;
  }

  @Override
  public Map<String, Reportable> getSubMetrics() {
    Map<String, Reportable> map = new HashMap<String, Reportable>();
    map.put(nodesMan.getName(), nodesMan);

    WALAckManager ack = getAckChecker();
    if (ack != null) {
      map.put(ack.getName(), ack);
    }

    map.put("jvmInfo", vmInfo);
    map.put("sysInfo", sysInfo);

    // TODO (jon) LivenessMan
    // TODO (jon) rpcMan

    return map;
  }

  public String getPhysicalNodeName() {
    return physicalNodeName;
  }

  public SystemInfo getSystemInfo() {
    return sysInfo;
  }

  public FlumeVMInfo getVMInfo() {
    return vmInfo;
  }
} public class NaiveFileWALDeco extends EventSinkDecorator<EventSink> {
  static final Logger LOG = LoggerFactory.getLogger(NaiveFileWALDeco.class);

  final WALManager walman;
  final RollTrigger trigger;
  final AckListener queuer;
  final EventSinkDecorator<EventSink> drainSink;
  final Context ctx;
  final long checkMs;
  final AckListener al;

  RollSink walProducer = null;
  EventSource walConsumer = null;
  DirectDriver walConsumerDriver = null;

  public NaiveFileWALDeco(Context ctx, EventSink s, final WALManager walman,
      RollTrigger t, AckListener al, long checkMs) {
    super(s);
    this.ctx = ctx;
    this.walman = walman;
    this.trigger = t;
    this.queuer = new AckListener.Empty();
    this.al = al;
    // this.drainSink = new LazyOpenDecorator<EventSink>(
    // new AckChecksumRegisterer<EventSink>(s, al));
    // this.drainSink = new AckChecksumRegisterer<EventSink>(
    // new LazyOpenDecorator<EventSink>(s), al);
    this.drainSink = new AckChecksumRegisterer<EventSink>(s, al);

    this.checkMs = checkMs;
  }

  /**
   * When this decorator encounters a correct checksum pair, it add registers
   * the checksume to the specified ack listener.
   */
  public static class AckChecksumRegisterer<S extends EventSink> extends
      EventSinkDecorator<S> {
    AckListener q;

    public AckChecksumRegisterer(S s, AckListener q) {
      super(s);
      this.q = q;
    }

    @Override
    public void append(Event e) throws IOException, InterruptedException {
      super.append(e);

      // check for and attempt to register interest in tag's checksum
      byte[] acktype = e.get(AckChecksumInjector.ATTR_ACK_TYPE);

      if (acktype == null) {
        // nothing more to do
        return;
      }

      byte[] acktag = e.get(AckChecksumInjector.ATTR_ACK_TAG);

      if (Arrays.equals(acktype, AckChecksumInjector.CHECKSUM_STOP)) {
        String k = new String(acktag);
        LOG.debug("Registering interest in checksum group called '" + k + "'");
        // Checksum stop marker: register interest
        q.end(k);
        return;
      }
    }
  }

  /**
   * TODO(jon): double check that the synchronization is appropriate here
   */
  @Override
  public synchronized void append(Event e) throws IOException,
      InterruptedException {
    Preconditions.checkNotNull(sink, "NaiveFileWALDeco was invalid!");
    Preconditions.checkState(isOpen.get(),
        "NaiveFileWALDeco not open for append");
    attemptToForwardException();
    walProducer.append(e);
  }

  /**
   * If the subthread has failed, fail by throwing the subthread's exception
   */
  void attemptToForwardException() throws IOException, InterruptedException {
    if (walConsumerDriver == null) {
      // if we fail during open before the driver is instantiated or started
      return;
    }
    Exception exn = walConsumerDriver.getException();
    if (exn != null) {
      if (exn instanceof IOException) {
        LOG.warn("wal consumer thread exited in error state: "
            + exn.getMessage());
        throw (IOException) exn;
      }
      if (exn instanceof InterruptedException) {
        LOG.warn("wal consumer thread exited in error due to interruption : "
            + exn.getMessage());
        throw (InterruptedException) exn;
      }
      LOG.error("Unexpected exception encountered! " + exn, exn);
    }
  }

  // This is going to go into the driver thread.
  void ensureClosedDrainDriver() throws IOException {
    LOG.debug("Ensuring wal consumer thread is done..");
    walman.stopDrains(); // signal wals source to stop (not blocking)
    if (walConsumerDriver == null) {
      // if we fail during open before the driver is instantiated or started
      return;
    }
    try {
      int maxNoProgressTime = 10;

      long evts = walConsumerDriver.appendCount;
      int count = 0;
      while (true) {
        if (walConsumerDriver.waitForAtLeastState(DriverState.IDLE, 500)) {
          // driver successfully completed
          LOG.debug(".. subthread to completed");
          break;
        }

        // driver still running, did we make progress?
        long evts2 = walConsumerDriver.appendCount;
        if (evts2 > evts) {
          // progress is being made, reset counts
          count = 0;
          evts = evts2;
          LOG.info("Closing wal log, subsink still making progress");
          continue;
        }

        // no progress.
        count++;
        LOG.info("Attempt " + count
            + " with no progress being made on wal consumer subsink");
        if (count >= maxNoProgressTime) {
          LOG.warn("WAL drain thread was not making progress, forcing close");
          walConsumerDriver.cancel();
          // walConsumerDriver.waitForAtLeastState(DriverState.IDLE, 5000);
          // break;
          continue; // should exit with atleast state ERROR
        }
      }

    } catch (InterruptedException e) {
      LOG.error("WAL drain thread interrupted", e);
    }
  }

  @Override
  public synchronized void close() throws IOException, InterruptedException {
    Preconditions.checkNotNull(sink,
        "Attmpted to close a null NaiveFileWALDeco");
    if (!isOpen.get()) {
      LOG.warn("Attempt to double close NaiveFileWALDeco");
      return;
    }

    LOG.debug("Closing NaiveFileWALDeco Source");
    if (walProducer != null) {
      walProducer.close(); // prevent new data from entering.
    } else {
      // TODO: how could this happen? things don't close in order?
      LOG.warn("input in naivefilewal deco was null!");
    }

    // Make sure subthread closed
    ensureClosedDrainDriver();

    isOpen.set(false);

    // This is throws an exception thrown by the subthread.
    attemptToForwardException();
    LOG.debug("Closed NaiveFileWALDeco");
  }

  @Override
  synchronized public void open() throws IOException, InterruptedException {
    Preconditions.checkNotNull(sink,
        "Attempted to open a null NaiveFileWALDeco subsink");
    LOG.debug("Opening NaiveFileWALDeco");
    walman.open();
    walman.recover();

    walProducer = walman.getAckingSink(ctx, trigger, queuer, checkMs);
    walConsumer = walman.getEventSource();

    // TODO (jon) catch exceptions here and close them before rethrowing
    walProducer.open();

    walConsumerDriver = new DirectDriver("naive file wal consumer",
        walConsumer, drainSink);
    walConsumerDriver.start();
    boolean success = walConsumerDriver.waitForAtLeastState(DriverState.ACTIVE,
        1000);
    if (!success) {
      walConsumerDriver.stop();
      attemptToForwardException();
    }
    LOG.debug("Opened NaiveFileWALDeco");
    Preconditions.checkNotNull(sink);
    Preconditions.checkState(!isOpen.get());
    isOpen.set(true);
  }

  /**
   * This should only be set before open is called.
   */
  public synchronized void setSink(EventSink sink) {
    Preconditions.checkState(!isOpen.get(),
        "setSink should not be called once NaiveFileWALDeco is open");
    this.sink = sink;
    // this.drainSink.setSink(new LazyOpenDecorator<EventSink>(
    // new AckChecksumRegisterer<EventSink>(sink, al)));
    // this.drainSink.setSink(new AckChecksumRegisterer<EventSink>(sink, al));
    // this.drainSink.setSink(new LazyOpenDecorator<EventSink>(sink));
    this.drainSink.setSink(sink);

  }

  public synchronized boolean rotate() throws InterruptedException {
    return walProducer.rotate();
  }

  public boolean isEmpty() {
    return walman.isEmpty();
  }

  /**
   * End to end ack version, with logical node context (to isolate wals.)
   * **/
  public static SinkDecoBuilder builderEndToEndDir() {
    return new SinkDecoBuilder() {
      @Override
      public EventSinkDecorator<EventSink> build(Context context,
          String... argv) {
        Preconditions.checkArgument(argv.length <= 3,
            "usage: ackedWriteAhead[(maxMillis,walnode,checkMs)]");
        FlumeConfiguration conf = FlumeConfiguration.get();
        long delayMillis = conf.getAgentLogMaxAge();

        if (argv.length >= 1) {
          delayMillis = Long.parseLong(argv[0]);
        }

        // TODO (jon) this will cause problems with multiple nodes in same JVM
        FlumeNode node = FlumeNode.getInstance();
        String walnode = context.getValue(LogicalNodeContext.C_LOGICAL);
        if (argv.length >= 2) {
          walnode = argv[1];
        }
        Preconditions.checkArgument(walnode != null,
            "Context does not have a logical node name");

        long checkMs = 250; // TODO replace with config var;
        if (argv.length >= 3) {
          checkMs = Long.parseLong(argv[2]);
        }

        // TODO (jon) this is going to be unsafe because it creates before open.
        // This needs to be pushed into the logic of the decorator
        WALManager walman = node.getAddWALManager(walnode);
        return new NaiveFileWALDeco(context, null, walman, new TimeTrigger(
            delayMillis), node.getAckChecker().getAgentAckQueuer(), checkMs);
      }
    };
  }

  @Override
  public String getName() {
    return "NaiveFileWAL";
  }

  @Deprecated
  @Override
  public ReportEvent getReport() {
    ReportEvent rpt = super.getReport();
    ReportEvent walRpt = walman.getMetrics();
    rpt.merge(walRpt);
    ReportEvent sinkReport = sink.getReport();
    rpt.hierarchicalMerge(getName(), sinkReport);
    return rpt;
  }

  @Override
  public ReportEvent getMetrics() {
    ReportEvent rpt = super.getMetrics();
    return rpt;
  }

  public Map<String, Reportable> getSubMetrics() {
    Map<String, Reportable> map = new HashMap<String, Reportable>();
    map.put(walman.getName(), walman);
    map.put("drainSink." + sink.getName(), sink);
    if (walConsumer != null) {
      // careful, drainSource can be null if deco not opened yet
      map.put("drainSource." + walConsumer.getName(), walConsumer);
    }
    return map;
  }

} public interface WALManager extends Reportable, WALCompletionNotifier {
  /**
   * These are the states a batch of events can be in.
   * 
   * IMPORT means external to the main flow of data.
   * 
   * WRITING means that there is some entity writing to the set of data.
   * 
   * LOGGED means that the data is durable on the node.
   * 
   * SENDING means that data is durable and is being sent somewhere downstream
   * 
   * SENT means that data has been received by the next downstream node.
   * 
   * E2EACKED means that data has been received and processed by the end point
   * node.
   * 
   * ERROR means that the set of data contains an error.
   */
  enum State {
    IMPORT, WRITING, LOGGED, SENDING, SENT, E2EACKED, ERROR
  };

  final public static String A_IMPORTED = "imported";
  final public static String A_WRITING = "writing";
  final public static String A_LOGGED = "logged";
  final public static String A_SENDING = "sending";
  final public static String A_SENT = "sent";
  final public static String A_ACKED = "acked";
  final public static String A_RETRY = "retry";
  final public static String A_ERROR = "error";
  final public static String A_RECOVERED = "recovered";
  final public static String A_IN_LOGGED = "loggedQ";
  final public static String A_IN_SENT = "sentQ";

  /**
   * Open the WAL manager to enable reads and writes.
   */
  public void open() throws IOException;

  /**
   * Closes the WAL manager for reads and writes
   */
  public void stopDrains() throws IOException;

  /**
   * Recover data that is not in E2EACKED or RROR state assuming that data
   * transmission has failed. This is called to recover durable data and retry
   * sends after a crash.
   */
  public void recover() throws IOException;

  /**
   * Get a new sink to write events to. Events are durably written before being
   * transmitted. The tagger creates a unique name for the batch of events.
   */
  public EventSink newWritingSink(Tagger t) throws IOException;

  /**
   * Gets an unacked batch. Read from the WAL by getting event sources from the
   * WAL manager.
   */
  public EventSource getUnackedSource() throws IOException, InterruptedException;

  /**
   * Get a logical sink that breaks stream of data into mini batches
   * transparently to the user. When the RollTrigger's predicate condition is
   * met, the rolling sink closes the previous sink and calls newWritingSink to
   * roll to a new writer.
   * 
   * An ackQueue listener reference is passed to provide callback hooks that are
   * called on different rolling sink transitions. (open, error, close).
   */
  public RollSink getAckingSink(Context ctx, RollTrigger t,
      final AckListener ackQueue, long checkMs) throws IOException;

  /**
   * Get a source that logically provides a single stream. This should call
   * getUnackedSource() underneath to get another event batch.
   */
  public EventSource getEventSource() throws IOException;

  /**
   * This imports "external data" into the WAL. Currently this is used for
   * testing.
   * 
   * TODO (jon) This interface is not quite right -- it should take a file and a
   * format as an arg. This will be revisited when we revisit the Log4J, Log4Net,
   * and Avro serialization integration.
   */
  public void importData() throws IOException;

  /**
   * Returns true if the WAL has no logged entries. This is useful for
   * completeness checking. (wait for after this is true before closing).
   */
  boolean isEmpty();

} public class Context {
  Map<String, Object> table = new HashMap<String, Object>();
  final Context parent;

  final public static Context EMPTY = new Context(null);

  public Context(Context parent) {
    this.parent = parent;
  }

  /**
   * This should only be used rarely. You probably want
   * LogicalNode.testingContext
   */
  public Context() {
    this.parent = EMPTY;
  }

  /**
   * Get the parent context
   */
  public Context getParent() {
    return this.parent;
  }

  /**
   * Get a value from the current node, otherwise go up to a larger scope and
   * try to get the value from there.
   */
  public String getValue(String arg) {
    Object val = table.get(arg);
    if (val != null) {
      return val.toString();
    }

    if (getParent() != null) {
      return getParent().getValue(arg);
    }
    return null; // no parent, and was null? return null
  }

  /**
   * Adds a value to the context assuming it is a string.
   */
  protected void putValue(String arg, String val) {
    table.put(arg, val);
  }

  /**
   * Adds an object value to the context.
   */
   protected <T> void putObj(String arg, T val) {
    table.put(arg, val);
  }

  /**
   * Gets an object value from the context.
   */
  public <T> T getObj(String arg, Class<T> clz) {
    Object val = table.get(arg);
    if (val == null && getParent() != null) {
      return getParent().getObj(arg, clz);
    }
    return clz.cast(val);
  }
} public class FlumeBuilder {
  static final Logger LOG = LoggerFactory.getLogger(FlumeBuilder.class);

  static SourceFactory srcFactory = new SourceFactoryImpl();
  static SinkFactory sinkFactory = new SinkFactoryImpl();

  enum ASTNODE {
    DEC, HEX, OCT, STRING, BOOL, FLOAT, // literals
    FUNC, // function node
    SINK, SOURCE, // sink or source
    KWARG, // kwarg support
    MULTI, DECO, BACKUP, ROLL, GEN, FAILCHAIN, // compound sinks
    NODE, // combination of sink and source
  };

  public static void setSourceFactory(SourceFactory srcFact) {
    Preconditions.checkNotNull(srcFact);
    srcFactory = srcFact;
  }

  public static void setSinkFactory(SinkFactory snkFact) {
    Preconditions.checkNotNull(snkFact);
    sinkFactory = snkFact;
  }

  /**
   * This hooks a particular string to the lexer. From there it creates a parser
   * that can be started from different entities. The lexer and language are
   * case sensitive.
   */
  static FlumeDeployParser getDeployParser(String s) {
    FlumeDeployLexer lexer = new FlumeDeployLexer(new ANTLRStringStream(s));
    CommonTokenStream tokens = new CommonTokenStream(lexer);
    return new FlumeDeployParser(tokens);
  }

  static CommonTree parse(String s) throws RecognitionException {
    return (CommonTree) getDeployParser(s).deflist().getTree();
  }

  static CommonTree parseHost(String s) throws RecognitionException {
    return (CommonTree) getDeployParser(s).host().getTree();
  }

  static CommonTree parseLiteral(String s) throws RecognitionException {
    return (CommonTree) getDeployParser(s).literal().getTree();
  }

  static CommonTree parseArg(String s) throws RecognitionException {
    return (CommonTree) getDeployParser(s).arg().getTree();
  }

  public static CommonTree parseSink(String s) throws RecognitionException {
    FlumeDeployParser parser = getDeployParser(s);
    CommonTree ast = (CommonTree) parser.sink().getTree();
    return ast;
  }

  public static CommonTree parseSource(String s) throws RecognitionException {
    return (CommonTree) getDeployParser(s).source().getTree();
  }

  /**
   * This is for reading and parsing out a full configuration from a file.
   */
  static CommonTree parseFile(String filename) throws IOException,
      RecognitionException {

    // Create a scanner and parser that reads from the input stream passed to us
    FlumeDeployLexer lexer = new FlumeDeployLexer(new ANTLRFileStream(filename));
    CommonTokenStream tokens = new CommonTokenStream(lexer);
    FlumeDeployParser parser = new FlumeDeployParser(tokens);

    return (CommonTree) parser.deflist().getTree();
  }

  /**
   * This is for reading and parsing out a full configuration from a file.
   */
  static CommonTree parseNodeFile(String filename) throws IOException,
      RecognitionException {

    // Create a scanner and parser that reads from the input stream passed to us
    FlumeDeployLexer lexer = new FlumeDeployLexer(new ANTLRFileStream(filename));
    CommonTokenStream tokens = new CommonTokenStream(lexer);
    FlumeDeployParser parser = new FlumeDeployParser(tokens);

    return (CommonTree) parser.connection().getTree();
  }

  public static Pair<EventSource, EventSink> buildNode(Context context, File f)
      throws IOException, RecognitionException, FlumeSpecException {
    CommonTree t = parseNodeFile(f.getCanonicalPath());
    if (t.getText() != "NODE") {
      throw new FlumeSpecException("fail, expected node but had "
          + t.toStringTree());
    }
    // String host = t.getChild(0).getText();
    CommonTree tsrc = (CommonTree) t.getChild(0);
    CommonTree tsnk = (CommonTree) t.getChild(1);

    return new Pair<EventSource, EventSink>(buildEventSource(context, tsrc),
        buildEventSink(context, tsnk, sinkFactory));
  }

  /**
   * This parses a aggregate configuration (name: src|snk; ...) and returns a
   * map from logical node name to a source sink pair. Context is required now
   * because a FlumeNode's PhysicalNode information may need to be passed in
   */
  @SuppressWarnings("unchecked")
  public static Map<String, Pair<String, String>> parseConf(Context ctx,
      String s) throws FlumeSpecException {
    try {
      CommonTree node = parse(s);
      Map<String, Pair<String, String>> cfg = new HashMap<String, Pair<String, String>>();
      for (CommonTree t : (List<CommonTree>) node.getChildren()) {
        // -1 is magic EOF value
        if (t.getType() == -1) {
          break; // I am done.
        }

        if (t.getText() != "NODE") {
          throw new FlumeSpecException("fail, expected node but had "
              + t.toStringTree());
        }

        if (t.getChildCount() != 3) {
          throw new FlumeSpecException(
              "fail, node didn't wasn't (name,src,snk): " + t.toStringTree());
        }
        String host = t.getChild(0).getText();
        CommonTree tsrc = (CommonTree) t.getChild(1);
        CommonTree tsnk = (CommonTree) t.getChild(2);

        Pair<String, String> p = new Pair<String, String>(
            FlumeSpecGen.genEventSource(tsrc), FlumeSpecGen.genEventSink(tsnk));
        cfg.put(host, p);
      }
      return cfg;
    } catch (RecognitionException re) {
      LOG.error("Failure to parse and instantiate sink: '" + s + "'", re);
      throw new FlumeSpecException(re.toString());
    }

  }

  @SuppressWarnings("unchecked")
  public static Map<String, Pair<EventSource, EventSink>> build(
      Context context, String s) throws FlumeSpecException {
    try {
      CommonTree node = parse(s);
      Map<String, Pair<EventSource, EventSink>> cfg = new HashMap<String, Pair<EventSource, EventSink>>();
      for (CommonTree t : (List<CommonTree>) node.getChildren()) {
        // -1 is magic EOF value
        if (t.getType() == -1) {
          break; // I am done.
        }
        if (t.getText() != "NODE") {
          throw new FlumeSpecException("fail, expected node but had "
              + t.toStringTree());
        }
        String host = t.getChild(0).getText();
        CommonTree tsrc = (CommonTree) t.getChild(1);
        CommonTree tsnk = (CommonTree) t.getChild(2);

        Pair<EventSource, EventSink> p = new Pair<EventSource, EventSink>(
            buildEventSource(context, tsrc), buildEventSink(context, tsnk,
                sinkFactory));
        cfg.put(host, p);
      }
      return cfg;
    } catch (RecognitionException re) {
      LOG.error("Failure to parse and instantiate sink: '" + s + "'", re);
      throw new FlumeSpecException(re.toString());
    }

  }

  /**
   * Build a flume source from a flume config specification.
   * 
   * This should only throw FlumeSpecExceptions (No illegal arg exceptions
   * anymore)
   */
  public static EventSource buildSource(Context ctx, String s)
      throws FlumeSpecException {
    try {
      CommonTree srcTree = parseSource(s);
      return buildEventSource(ctx, srcTree);
    } catch (RecognitionException re) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", re);
      throw new FlumeSpecException(re.toString());
    } catch (NumberFormatException nfe) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", nfe);
      throw new FlumeSpecException(nfe.getMessage());
    } catch (IllegalArgumentException iae) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", iae);
      throw new FlumeSpecException(iae.getMessage());
    } catch (RuntimeRecognitionException re) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", re);
      throw new FlumeSpecException(re.getMessage());
    }
  }

  /**
   * Build a flume sink from a flume config specification. Sinks can be much
   * more complicated than sources.
   * 
   * This should only throw FlumeSpecExceptions (No illegal arg exceptions
   * anymore)
   */
  public static EventSink buildSink(Context context, String s)
      throws FlumeSpecException {
    try {
      CommonTree snkTree = parseSink(s);
      return buildEventSink(context, snkTree, sinkFactory);
    } catch (RecognitionException re) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", re);
      throw new FlumeSpecException(re.toString());
    } catch (NumberFormatException nfe) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", nfe);
      throw new FlumeSpecException(nfe.getMessage());
    } catch (IllegalArgumentException iae) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", iae);
      throw new FlumeSpecException(iae.getMessage());
    } catch (RuntimeRecognitionException re) {
      LOG.debug("Failure to parse and instantiate sink: '" + s + "'", re);
      throw new FlumeSpecException(re.getMessage());
    }
  }

  @Deprecated
  public static EventSink buildSink(String s) throws FlumeSpecException {
    return buildSink(new Context(), s);
  }

  /**
   * Formats the three pieces to a node configuration back into a single
   * parsable line
   */
  public static String toLine(String name, String src, String snk) {
    return name + " : " + src + " | " + snk + ";";
  }

  /**
   * All of factories expect all string arguments.
   * 
   * Numbers are expected to be in decimal format.
   * 
   * If it is of type KWARG, it returns null
   * 
   * TODO (jon) move to builders, or allow builders to take Integers/Booleans as
   * well as Strings
   */
  public static Object buildSimpleArg(CommonTree t) throws FlumeSpecException {
    ASTNODE type = ASTNODE.valueOf(t.getText()); // convert to enum
    switch (type) {
    case HEX:
      String hex = t.getChild(0).getText();
      Preconditions.checkArgument(hex.startsWith("0x")); // bad parser if this
      // happens
      hex = hex.substring(2);
      Long i = Long.parseLong(hex, 16); // use base 16 radix
      return i.toString();
    case DEC:
      return t.getChild(0).getText();
    case BOOL:
      return t.getChild(0).getText();
    case OCT:
      String oct = t.getChild(0).getText();
      // bad parser if these happen
      Preconditions.checkArgument(oct.startsWith("0"));
      Preconditions.checkArgument(!oct.startsWith("0x"));
      Long i2 = Long.parseLong(oct, 8); // us base 16 radix
      return i2.toString();
    case FLOAT:
      return t.getChild(0).getText();
    case STRING:
      String str = t.getChild(0).getText();
      Preconditions.checkArgument(str.startsWith("\"") && str.endsWith("\""));
      str = str.substring(1, str.length() - 1);
      return StringEscapeUtils.unescapeJava(str);
    case FUNC:
      String name = t.getChild(0).getText();
      List<Object> args = new ArrayList<Object>();
      int children = t.getChildCount();
      for (int j = 1; j < children; j++) {
        args.add(buildSimpleArg((CommonTree) t.getChild(j)));
      }
      FunctionSpec funSpec = new FunctionSpec(name, args.toArray());
      return funSpec;
    case KWARG:
      return null;
    default:
      throw new FlumeSpecException("Not a node of literal type: "
          + t.toStringTree());
    }
  }

  /**
   * Container class for holding function arguments
   */
  public static class FunctionSpec {
    String name;
    Object[] args;

    FunctionSpec(String name, Object... args) {
      this.name = name;
      this.args = args;
    }

    public String getName() {
      return name;
    }

    public Object[] getArgs() {
      return args;
    }

    public String toString() {
      return name + " " + Arrays.toString(args);
    }
  }

  public static OutputFormat createFormat(FormatFactory ff, Object o ) throws FlumeSpecException {
    if (o == null) {
      // return the default.
      return ff.createOutputFormat(null);
    }
    if (o instanceof FunctionSpec) {
      FunctionSpec fs = (FunctionSpec)o;
      return ff.createOutputFormat(fs.getName(), fs.getArgs());
    }

    LOG.warn("Deprecated syntax: Expected a format spec but instead "
        + "had a (" + o.getClass().getSimpleName() + ") "
        + o.toString());
    return ff.getOutputFormat(o.toString());
  }


  public static Pair<String, CommonTree> buildKWArg(CommonTree t) {
    ASTNODE type = ASTNODE.valueOf(t.getText()); // convert to enum
    Preconditions.checkArgument("KWARG".equals(type.toString()));
    String kw = t.getChild(0).getText();
    CommonTree arg = (CommonTree) t.getChild(1);
    return new Pair<String, CommonTree>(kw, arg);
  }

  /**
   * This method populates the id, argv,and kwargs into ctx needed to build
   * source/sink/deco name
   * 
   * @param t
   * @param ctx
   * @return
   * @throws FlumeSpecException
   */
  @SuppressWarnings("unchecked")
  static Pair<String, List<Object>> handleArgs(CommonTree t, Context ctx)
      throws FlumeSpecException {
    List<CommonTree> children = (List<CommonTree>) new ArrayList<CommonTree>(
        t.getChildren());
    String sinkType = children.remove(0).getText();
    List<Object> args = new ArrayList<Object>();
    for (CommonTree tr : children) {
      Object arg = buildSimpleArg(tr);
      if (arg != null) {
        args.add(arg);
      } else {
        Pair<String, CommonTree> kwarg = buildKWArg(tr);
        ctx.putObj(kwarg.getLeft(), buildSimpleArg(kwarg.getRight()));
      }
    }
    return new Pair<String, List<Object>>(sinkType, args);
  }

  static EventSource buildEventSource(Context context, CommonTree t)
      throws FlumeSpecException {
    ASTNODE type = ASTNODE.valueOf(t.getText()); // convert to enum
    switch (type) {
    case SOURCE: {
      Context ctx = new Context(context);
      Pair<String, List<Object>> idArgs = handleArgs(t, ctx);
      String sourceType = idArgs.getLeft();
      List<Object> args = idArgs.getRight();

      EventSource src = srcFactory
          .createSource(ctx, sourceType, args.toArray());
      if (src == null) {
        throw new FlumeIdException("Invalid source: "
            + FlumeSpecGen.genEventSource(t));
      }
      return src;
    }
    default:
      throw new FlumeSpecException("bad parse tree! Expected source but got "
          + t.toStringTree());
    }
  }

  @SuppressWarnings("unchecked")
  static EventSink buildEventSink(Context context, CommonTree t,
      SinkFactory sinkFactory) throws FlumeSpecException {
    ASTNODE type = ASTNODE.valueOf(t.getText()); // convert to enum
    switch (type) {
    case SINK: {

      Context ctx = new Context(context);
      Pair<String, List<Object>> idargs = handleArgs(t, ctx);
      String sinkType = idargs.getLeft();
      List<Object> args = idargs.getRight();

      EventSink snk = sinkFactory.createSink(ctx, sinkType, args.toArray());

      if (snk == null) {
        throw new FlumeIdException("Invalid sink: "
            + FlumeSpecGen.genEventSink(t));
      }
      return snk;
    }
    case MULTI: {
      List<CommonTree> elems = (List<CommonTree>) t.getChildren();
      List<EventSink> snks = new ArrayList<EventSink>();
      try {
        for (CommonTree tr : elems) {
          EventSink s = buildEventSink(context, tr, sinkFactory);
          snks.add(s);
        }
        FanOutSink<EventSink> sink = new FanOutSink<EventSink>(snks);
        return sink;
      } catch (FlumeSpecException ife) {
        // TODO (jon) do something if there was an intermediate failure
        throw ife;
      }
    }
    case DECO: {
      List<CommonTree> decoNodes = (List<CommonTree>) t.getChildren();
      Preconditions.checkArgument(decoNodes.size() <= 2,
          "Only supports one or no decorators per expression");
      // no decorators
      if (decoNodes.size() == 1) {
        CommonTree snkt = decoNodes.get(0);
        try {
          EventSink singleSnk = buildEventSink(context, snkt, sinkFactory);
          return singleSnk;
        } catch (FlumeSpecException ife) {
          throw ife;
        }
      }

      // single decorator
      CommonTree deco = decoNodes.get(0);
      CommonTree decoSnk = decoNodes.get(1);
      EventSinkDecorator<EventSink> decoSink = buildEventSinkDecorator(context,
          deco);
      try {
        EventSink dsnk = buildEventSink(context, decoSnk, sinkFactory);
        decoSink.setSink(dsnk);
        return decoSink;
      } catch (FlumeSpecException ife) {
        // TODO (jon) need to cleanup after mainsink if this failed.
        throw ife;
      }
    }

    case BACKUP: {
      List<CommonTree> backupNodes = (List<CommonTree>) t.getChildren();
      Preconditions.checkArgument(backupNodes.size() == 2,
          "Only supports two retry nodes per failover expression");
      CommonTree main = backupNodes.get(0);
      CommonTree backup = backupNodes.get(1);
      try {
        EventSink mainSink = buildEventSink(context, main, sinkFactory);
        EventSink backupSink = buildEventSink(context, backup, sinkFactory);
        return new BackOffFailOverSink(mainSink, backupSink);
      } catch (FlumeSpecException ife) {
        LOG.error("Failed to build Failover sink", ife);
        throw ife;
      }
    }

    case ROLL: {
      List<CommonTree> rollArgs = (List<CommonTree>) t.getChildren();
      try {
        Preconditions.checkArgument(rollArgs.size() >= 2, "bad parse tree! "
            + t.toStringTree() + "roll requires at least two arguments");
        SinkBuilder rollBuilder = RollSink.builder();
        Context ctx = new Context(context); // new scope
        List<Object> args = new ArrayList<Object>();

        // first is the sub sink spec
        CommonTree ctbody = rollArgs.get(0);
        String body= FlumeSpecGen.genEventSink(ctbody);
        args.add(body);

        // all others are args
        for (int i =1; i< rollArgs.size(); i++) {
          CommonTree tr = rollArgs.get(i);
          Object arg = buildSimpleArg(tr);
          if (arg != null) {
            args.add(arg);
          } else {
            Pair<String, CommonTree> kwarg = buildKWArg(tr);
            ctx.putObj(kwarg.getLeft(), buildSimpleArg(kwarg.getRight()));
          }
        }
        return rollBuilder.create(ctx, args.toArray());
      } catch (IllegalArgumentException iae) {
        throw new FlumeSpecException(iae.getMessage());
      }
    }

    case GEN: {
      List<CommonTree> collArgs = (List<CommonTree>) t.getChildren();
      try {
        Preconditions.checkArgument(collArgs.size() >= 2, "bad parse tree! "
            + t.toStringTree() + " generator takes at least 2 arguments");
        String genType = collArgs.get(0).getText();
        if (!"collector".equals(genType)) {
          throw new FlumeSpecException(
              "currently only handle 'collector' gen sinks");
        }
        CommonTree ctbody = collArgs.remove(1); // remove subsink.
        String body = FlumeSpecGen.genEventSink(ctbody);

        Context ctx = new Context(context);
        Pair<String, List<Object>> idArgs = handleArgs(t, ctx);
        String sourceType = idArgs.getLeft();
        List<Object> args = idArgs.getRight();
        args.add(0, body);

        // TODO replace with Generator Sink lookup
        Preconditions.checkArgument("collector".equals(sourceType));
        SinkBuilder builder = CollectorSink.builder();

        return builder.create(ctx, args.toArray());

      } catch (IllegalArgumentException iae) {
        throw new FlumeSpecException(iae.getMessage());
      }

    }

    case FAILCHAIN: {
      // TODO (jon) This is no longer necessary with the substitution mechanisms
      // found in the translators
      List<CommonTree> rollArgs = (List<CommonTree>) t.getChildren();
      Preconditions.checkArgument(rollArgs.size() >= 2);
      CommonTree ctbody = rollArgs.get(0);
      List<String> rargs = new ArrayList<String>(rollArgs.size() - 1);
      boolean first = true;
      for (CommonTree ct : rollArgs) {
        if (first) {
          first = false;
          continue;
        }
        // assumes this is a STRING
        rargs.add(buildSimpleArg(ct).toString());
      }
      String body = FlumeSpecGen.genEventSink(ctbody);
      FlumeConfiguration conf = FlumeConfiguration.get();
      FailoverChainSink failchain = new FailoverChainSink(context, body, rargs,
          conf.getFailoverInitialBackoff(), conf.getFailoverMaxSingleBackoff());
      return failchain;
    }
      // TODO (jon) new feature: handle pattern match splitter
      // case MATCH:

    default:
      throw new FlumeSpecException("bad parse tree! expected sink but got "
          + t.toStringTree());
    }
  }

  static EventSinkDecorator<EventSink> buildEventSinkDecorator(Context context,
      CommonTree t) throws FlumeSpecException {
    Context ctx = new Context(context);
    Pair<String, List<Object>> idArgs = handleArgs(t, ctx);
    String sinkType = idArgs.getLeft();
    List<Object> args = idArgs.getRight();
    EventSinkDecorator<EventSink> deco = sinkFactory.createDecorator(ctx,
        sinkType, args.toArray());
    if (deco == null) {
      throw new FlumeIdException("Invalid sink decorator: "
          + FlumeSpecGen.genEventSinkDecorator(t));
    }
    return deco;
  }

  /**
   * Returns an unmodifiable set containing the sinks this builder supports
   */
  public static Set<String> getSinkNames() {
    return sinkFactory.getSinkNames();
  }

  /**
   * Return an unmodifiable set containing the decorators this builder supports
   */
  public static Set<String> getDecoratorNames() {
    return sinkFactory.getDecoratorNames();
  }

  /**
   * Return an unmodifiable set containing the sources this builder supports
   */
  public static Set<String> getSourceNames() {
    return srcFactory.getSourceNames();
  }
} public class FlumeConfiguration extends Configuration {
  protected static final Logger LOG = LoggerFactory
      .getLogger(FlumeConfiguration.class);
  private static FlumeConfiguration singleton;

  /**
   * Returns the 'FLUME_HOME' location. Taken in order of precedence:
   * 
   * - Java system property 'flume.home'
   * 
   * - $FLUME_HOME in the environment.
   * 
   * - null if neither of these are set.
   */
  public static String getFlumeHome() {
    String flumeHome = System.getProperty("flume.home");
    if (null == flumeHome) {
      flumeHome = System.getenv("FLUME_HOME");
    }

    if (null == flumeHome) {
      LOG.warn("-Dflume.home and $FLUME_HOME both unset");
    }

    return flumeHome;
  }

  /**
   * Returns the 'FLUME_CONF_DIR' location. Taken in order of precedence:
   * 
   * - Java system property 'flume.conf.dir'
   * 
   * - $FLUME_CONF_DIR in the environment
   * 
   * - getFlumeHome()/conf
   * 
   * - ./conf
   */
  public static String getFlumeConfDir() {
    String flumeConfDir = System.getProperty("flume.conf.dir");
    if (null == flumeConfDir) {
      flumeConfDir = System.getenv("FLUME_CONF_DIR");
    }

    if (null == flumeConfDir) {
      String flumeHome = getFlumeHome();
      if (null != flumeHome) {
        flumeConfDir = new File(flumeHome, "conf").toString();
      } else {
        flumeConfDir = "./conf";
      }
    }

    return flumeConfDir;
  }

  public synchronized static FlumeConfiguration get() {
    if (singleton == null)
      singleton = new FlumeConfiguration();
    return singleton;
  }

  protected FlumeConfiguration() {
    this(true);
  }

  /**
   * Resets the singleton configuration to empty and returns it. Should be used
   * for testing only - this is used to isolate different test configurations.
   */
  public synchronized static FlumeConfiguration createTestableConfiguration() {
    singleton = new FlumeConfiguration(false);
    return get();
  }

  protected FlumeConfiguration(boolean loadDefaults) {
    super();
    if (loadDefaults) {
      Path conf = new Path(getFlumeConfDir());
      LOG.info("Loading configurations from " + conf);
      super.addResource(new Path(conf, "flume-conf.xml"));
      super.addResource(new Path(conf, "flume-site.xml"));
    }
  }

  // Default values
  static public final int DEFAULT_HEARTBEAT_PORT = 35872;
  static public final int DEFAULT_GOSSIP_PORT = 57890;
  static public final int DEFAULT_ADMIN_PORT = 35873;
  static public final int DEFAULT_HTTP_PORT = 35871;
  static public final int DEFAULT_REPORT_SERVER_PORT = 45678;

  public static final int DEFAULT_ZK_CLIENT_PORT = 3181;
  public static final int DEFAULT_ZK_SERVER_QUORUM_PORT = 3182;
  public static final int DEFAULT_ZK_SERVER_ELECTION_PORT = 3183;

  // Default sink / source variables
  static public final int DEFAULT_SCRIBE_SOURCE_PORT = 1463;

  // Watch dog parameters
  public final static String WATCHDOG_MAX_RESTARTS = "watchdog.restarts.max";

  // Agent parameters
  public final static String AGENT_LOG_DIR_NEW = "flume.agent.logdir";
  public final static String AGENT_LOG_MAX_AGE = "flume.agent.logdir.maxage";
  public static final String AGENT_LOG_ACKED_RETRANSMIT_AGE = "flume.agent.logdir.retransmit";

  public final static String NODE_STATUS_PORT = "flume.node.status.port";
  public static final String AGENT_FAILOVER_INITIAL_BACKOFF = "flume.agent.failover.backoff.initial";
  public static final String AGENT_FAILOVER_MAX_BACKOFF = "flume.agent.failover.backoff.max";
  public static final String AGENT_FAILOVER_MAX_CUMULATIVE_BACKOFF = "flume.agent.failover.backoff.cumulativemax";
  public static final String AGENT_HEARTBEAT_BACKOFF = "flume.agent.heartbeat.backoff";
  public static final String AGENT_MEMTHRESHOLD = "flume.agent.mem.threshold";
  public static final String AGENT_MULTIMASTER_MAXRETRIES = "flume.agent.multimaster.maxretries";
  public static final String AGENT_MULTIMASTER_RETRYBACKOFF = "flume.agent.multimaster.retrybackoff";
  public static final String WAL_OUTPUT_BUFFER = "flume.node.wal.output.buffered";

  // Flow options
  public static final String DEFAULT_FLOW_NAME = "flume.flow.default.name";

  // Sink/source default options
  public static final String POLLER_QUEUESIZE = "flume.poller.queuesize";
  public static final String THRIFT_QUEUESIZE = "flume.thrift.queuesize";
  public static final String THRIFT_CLOSE_MAX_SLEEP = "flume.thrift.close.maxsleep";
  public static final String THRIFT_SOCKET_TIMEOUT_MS = "flume.thrift.socket.timeout.ms";
  public static final String INSISTENTOPEN_INIT_BACKOFF = "flume.inisistentOpen.init.backoff";
  public static final String HISTORY_DEFAULTPERIOD = "flume.countHistory.period";
  public static final String HISTORY_MAXLENGTH = "flume.history.maxlength";
  public static final String TAIL_POLLPERIOD = "flume.tail.pollperiod";
  public static final String EVENT_RPC_TYPE = "flume.event.rpc";

  // Collector parameters
  public final static String COLLECTOR_EVENT_HOST = "flume.collector.event.host";
  public final static String COLLECTOR_EVENT_PORT = "flume.collector.event.port";
  public static final String COLLECTOR_DFS_DIR = "flume.collector.dfs.dir";
  public static final String COLLECTOR_ROLL_MILLIS = "flume.collector.roll.millis";
  public static final String COLLECTOR_OUTPUT_FORMAT = "flume.collector.output.format";
  public static final String COLLECTOR_DFS_COMPRESS_CODEC = "flume.collector.dfs.compress.codec";

  // TODO(henry) move these to flume.master - they now tell the master which
  // interface / port to start up on
  public static final String MASTER_HTTP_PORT = "flume.master.http.port";
  public static final String MASTER_HEARTBEAT_PORT = "flume.master.heartbeat.port";
  public static final String MASTER_HEARTBEAT_SERVERS = "flume.master.heartbeat.servers";
  public static final String MASTER_HEARBEAT_RPC = "flume.master.heartbeat.rpc";

  public static final String CONFIG_HEARTBEAT_PERIOD = "flume.config.heartbeat.period";
  public static final String MASTER_HEARTBEAT_MAX_MISSED = "flume.config.heartbeat.missed.max";
  public static final String NODE_HEARTBEAT_BACKOFF_LIMIT = "flume.node.heartbeat.backoff.ceiling";
  public static final String NODE_HTTP_AUTOFINDPORT = "flume.node.http.autofindport";
  public static final String NODE_CLOSE_TIMEOUT = "flume.node.close.timeout";
  public static final String CONFIG_ADMIN_PORT = "flume.config.admin.port";
  public static final String REPORT_SERVER_PORT = "flume.report.server.port";
  public static final String REPORT_SERVER_RPC_TYPE = "flume.report.server.rpc.type";
  public static final String MASTER_SAVEFILE = "flume.master.savefile";
  public static final String MASTER_SAVEFILE_AUTOLOAD = "flume.master.savefile.autoload";

  // reporter parameters
  public static final String REPORTER_POLLER_PERIOD = "flume.reporter.poller.period";

  // security parameters
  public static final String SECURITY_KERBEROS_PRINCIPAL = "flume.security.kerberos.principal";
  public static final String SECURITY_KERBEROS_KEYTAB = "flume.security.kerberos.keytab";

  public static final String FLURKER_ENCODING = "flume.irc.encoding";

  public static final String TWITTER_STREAM_URL = "flume.twitter.url";
  public static final String TWITTER_USER = "flume.twitter.username";
  public static final String TWITTER_PW = "flume.twitter.password";

  public static final String SCRIBE_SOURCE_PORT = "flume.scribe.source.port";

  public static final String WEBAPPS_PATH = "flume.webapps.path";

  // 80% of the time, this is all you need to set
  public static final String MASTER_SERVERS = "flume.master.servers";

  public static final String MASTER_STORE = "flume.master.store";
  public static final String MASTER_SERVER_ID = "flume.master.serverid";

  // List of masters and their gossip ports - used to override MASTER_SERVERS
  // where necessary
  public static final String MASTER_GOSSIP_SERVERS = "flume.master.gossip.servers";
  public static final String MASTER_GOSSIP_PERIOD_MS = "flume.master.gossip.period";
  public static final String MASTER_GOSSIP_MAXAGE_MS = "flume.master.gossip.maxage";
  public static final String MASTER_GOSSIP_PORT = "flume.master.gossip.port";

  // ZooKeeper bits and pieces
  public static final String MASTER_ZK_LOGDIR = "flume.master.zk.logdir";
  public static final String MASTER_ZK_CLIENT_PORT = "flume.master.zk.client.port";
  public static final String MASTER_ZK_SERVER_QUORUM_PORT = "flume.master.zk.server.quorum.port";
  public static final String MASTER_ZK_SERVER_ELECTION_PORT = "flume.master.zk.server.election.port";
  public static final String MASTER_ZK_SERVERS = "flume.master.zk.servers";
  public static final String MASTER_ZK_USE_EXTERNAL = "flume.master.zk.use.external";

  // These are configuration variables that are passed straight to ZK
  // 99% of installations shouldn't need most of them
  public static final String ZK_TICK_TIME = "flume.zk.ticktime";
  public static final String ZK_INIT_LIMIT = "flume.zk.initlimit";
  public static final String ZK_SYNC_LIMIT = "flume.zk.synclimit";

  public static final String EVENT_MAX_SIZE = "flume.event.max.size.bytes";

  public static final String GANGLIA_SERVERS = "flume.ganglia.servers";

  public static final String HIVE_HOST = "flume.hive.host";
  public static final String HIVE_PORT = "flume.hive.port";
  public static final String HIVE_USER = "flume.hive.user";
  public static final String HIVE_PW = "flume.hive.userpw";

  public static final String PLUGIN_CLASSES = "flume.plugin.classes";
  public static final String OUTPUT_FORMAT_PLUGIN_CLASSES = "flume.plugin.outputformat.classes";

  // Options for RPC type
  public static final String RPC_TYPE_THRIFT = "THRIFT";
  public static final String RPC_TYPE_AVRO = "AVRO";

  // Web app options
  public static final String WEBAPP_ROOT_NODE = "flume.node.webapp.root";
  public static final String WEBAPP_ROOT_MASTER = "flume.master.webapp.root";

  /**
   * Returns true if there is more than one server in MASTER_SERVERS.
   */
  public boolean getMasterIsDistributed() {
    return getMasterServers().split(",").length > 1;
  }

  /**
   * Returns the kind of store that the master should use. Using "memory" in
   * conjunction with a distributed setting is an error and the client should
   * pick up on that.
   */
  public String getMasterStore() {
    return get(MASTER_STORE, "zookeeper");
  }

  /**
   * Get a list in host:clientport:serverport format of servers to connect to
   * for ZK. If MASTER_ZK_SERVERS is set, return that. Otherwise, return the
   * servers from MASTER_SERVERS with MASTER_ZK_CLIENT_PORT and
   * MASTER_ZK_SERVER_PORT as default values.
   */
  public String getMasterZKServers() {
    String servers = get(MASTER_ZK_SERVERS, null);
    if (servers != null) {
      return servers;
    }

    String[] hosts = getMasterServers().split(",");
    int clientport = getMasterZKClientPort();
    int quorumport = getMasterZKServerQuorumPort();
    int electionport = getMasterZKServerElectionPort();
    List<String> l = Arrays.asList(hosts);
    Iterator<String> iter = l.iterator();
    StringBuilder builder = new StringBuilder();
    while (iter.hasNext()) {
      builder.append(iter.next() + ":" + clientport + ":" + quorumport + ":"
          + electionport);
      if (iter.hasNext()) {
        builder.append(',');
      }
    }
    return builder.toString();
  }

  /**
   * Returns the string that a client should use when connecting to ZooKeeper.
   * List of hostname:clientport comma-separated pairs. If no servers
   * configured, will be empty.
   */
  public String getMasterZKConnectString() {
    String servers = getMasterZKServers();
    String[] hosts = servers.split(",");
    StringBuilder builder = new StringBuilder();
    for (int i = 0; i < hosts.length; ++i) {
      hosts[i] = hosts[i].trim();

      String[] parts = hosts[i].split(":");
      builder.append(parts[0] + ":" + parts[1]);
      if (i < hosts.length - 1) {
        builder.append(",");
      }
    }
    return builder.toString();
  }

  /**
   * The client port that the in-process ZK starts on.
   * 
   * If it is set, return that.
   * 
   * If MASTER_ZK_SERVERS is set, then we return the clientport in that string
   * corresponding to our serverid.
   * 
   * Otherwise return DEFAULT_ZK_CLIENT_PORT.
   * 
   */
  public int getMasterZKClientPort() {
    String clientport = get(MASTER_ZK_CLIENT_PORT, null);
    if (clientport != null) {
      return Integer.parseInt(clientport);
    }

    // clientport is not set, try and guess from MASTER_ZK_SERVERS if it's set
    // We can't call getMasterZKServers because that might call into this method
    // itself...
    String servers = get(MASTER_ZK_SERVERS, null);
    if (servers == null) {
      return DEFAULT_ZK_CLIENT_PORT;
    }

    // MASTER_ZK_SERVERS is set - split it and guess at our client port
    List<String> serverList = Arrays.asList(servers.split(","));
    int serverid = getMasterServerId();
    Preconditions.checkState(serverid < serverList.size(),
        "Serverid is out of range: " + serverid);

    String[] server = serverList.get(serverid).split(":");
    Preconditions.checkState(server.length == 4, "Server spec "
        + serverList.get(serverid) + " is ill-formed");
    return Integer.parseInt(server[1].trim());
  }

  /**
   * The server quorum port that the in-process ZK starts on.
   * 
   * If it is set, return that.
   * 
   * If MASTER_ZK_SERVERS is set, then we return the quorumport in that string
   * corresponding to our serverid.
   * 
   * Otherwise return DEFAULT_ZK_SERVER_QUORUM_PORT, the default.
   * 
   */
  public int getMasterZKServerQuorumPort() {
    String quorumport = get(MASTER_ZK_SERVER_QUORUM_PORT, null);
    if (quorumport != null) {
      return Integer.parseInt(quorumport);
    }

    // quorumport is not set, try and guess from MASTER_ZK_SERVERS if it's set
    // We can't call getMasterZKServers because that might call into this method
    // itself...
    String servers = get(MASTER_ZK_SERVERS, null);
    if (servers == null) {
      return DEFAULT_ZK_SERVER_QUORUM_PORT;
    }

    // MASTER_ZK_SERVERS is set - split it and guess at our server port
    List<String> serverList = Arrays.asList(servers.split(","));
    int serverid = getMasterServerId();
    Preconditions.checkState(serverid < serverList.size(),
        "Serverid is out of range: " + serverid);

    String[] server = serverList.get(serverid).split(":");
    Preconditions.checkState(server.length == 4, "Server spec "
        + serverList.get(serverid) + " is ill-formed");
    return Integer.parseInt(server[2].trim());
  }

  /**
   * The server election port that the in-process ZK starts on.
   * 
   * If it is set, return that.
   * 
   * If MASTER_ZK_SERVERS is set, then we return the electionport in that string
   * corresponding to our serverid.
   * 
   * Otherwise return DEFAULT_ZK_SERVER_ELECTION_PORT, the default.
   * 
   */
  public int getMasterZKServerElectionPort() {
    String electionport = get(MASTER_ZK_SERVER_ELECTION_PORT, null);
    if (electionport != null) {
      return Integer.parseInt(electionport);
    }

    // electionport is not set, try and guess from MASTER_ZK_SERVERS if it's set
    // We can't call getMasterZKServers because that might call into this method
    // itself...
    String servers = get(MASTER_ZK_SERVERS, null);
    if (servers == null) {
      return DEFAULT_ZK_SERVER_ELECTION_PORT;
    }

    // MASTER_ZK_SERVERS is set - split it and guess at our server port
    List<String> serverList = Arrays.asList(servers.split(","));
    int serverid = getMasterServerId();
    Preconditions.checkState(serverid < serverList.size(),
        "Serverid is out of range: " + serverid);

    String[] server = serverList.get(serverid).split(":");
    Preconditions.checkState(server.length == 4, "Server spec "
        + serverList.get(serverid) + " is ill-formed");
    return Integer.parseInt(server[3].trim());
  }

  public boolean getMasterZKUseExternal() {
    return getBoolean(MASTER_ZK_USE_EXTERNAL, false);
  }

  /**
   * Where should ZK server store its persistent logs? (Shouldn't be in tmp!)
   */
  public String getMasterZKLogDir() {
    return get(MASTER_ZK_LOGDIR, "/tmp/flume-${user.name}/master/zk");
  }

  /**
   * The amount of time, in ms, that a 'tick' takes in ZK.
   */
  public int getMasterZKTickTime() {
    return getInt(ZK_TICK_TIME, 2000);
  }

  /**
   * Amount of time, in ticks, to allow followers to connect and sync to a
   * leader.
   * 
   * @return
   */
  public int getMasterZKInitLimit() {
    return getInt(ZK_INIT_LIMIT, 10);
  }

  /**
   * Amount of time, in ticks, that ZK followers can take to sync up with the
   * leader.
   */
  public int getMasterZKSyncLimit() {
    return getInt(ZK_SYNC_LIMIT, 5);
  }

  /**
   * For distributed ZK, we need to know which server we are.
   */
  public int getMasterServerId() {
    return getInt(MASTER_SERVER_ID, 0);
  }

  public String getAgentLogsDir() {
    return get(AGENT_LOG_DIR_NEW, "/tmp/flume-${user.name}/agent");
  }

  public long getAgentLogMaxAge() {
    return getLong(AGENT_LOG_MAX_AGE, 10000);
  }

  /**
   * Maximum number of application restarts per minute
   */
  public int getMaxRestartsPerMin() {
    return getInt(WATCHDOG_MAX_RESTARTS, 5);
  }

  public int getNodeStatusPort() {
    return getInt(NODE_STATUS_PORT, 35862);
  }

  public long getAgentAckedRetransmit() {
    return getLong(AGENT_LOG_ACKED_RETRANSMIT_AGE, 60000);
  }

  /**
   * Heap memory usage threshold (between 0.0 and 1.0) after a forced gc.
   */
  public float getAgentMemoryThreshold() {
    return getFloat(AGENT_MEMTHRESHOLD, 0.95F);
  }

  /**
   * Maximum number of times to retry connecting to all masters.
   */
  public int getAgentMultimasterMaxRetries() {
    return getInt(AGENT_MULTIMASTER_MAXRETRIES, 12);
  }

  /**
   * Amount of time to backoff before attempting to re-contact a master when all
   * fail.
   */
  public int getAgentMultimasterRetryBackoff() {
    return getInt(AGENT_MULTIMASTER_RETRYBACKOFF, 5000);
  }

  public long getFailoverInitialBackoff() {
    return getLong(AGENT_FAILOVER_INITIAL_BACKOFF, 1000);
  }

  public long getFailoverMaxSingleBackoff() {
    return getLong(AGENT_FAILOVER_MAX_BACKOFF, 60000);
  }

  public long getFailoverMaxCumulativeBackoff() {
    return getLong(AGENT_FAILOVER_MAX_CUMULATIVE_BACKOFF, Integer.MAX_VALUE);
  }

  /**
   * Max queue size for a poller source.
   */
  public int getPollerQueueSize() {
    return getInt(POLLER_QUEUESIZE, 100);
  }

  /**
   * Max queue size for a poller source.
   */
  public int getThriftQueueSize() {
    return getInt(THRIFT_QUEUESIZE, 1000);
  }

  public int getThriftSocketTimeoutMs() {
    return getInt(THRIFT_SOCKET_TIMEOUT_MS, 10000);
  }

  /**
   * Initial backoff in mills after a failed open attempt in an insistentOpen
   * decorator
   */
  public long getInsistentOpenInitBackoff() {
    return getLong(INSISTENTOPEN_INIT_BACKOFF, 1000);
  }

  /**
   * Period for to reset counters in millis
   */
  public long getHistoryDefaultPeriod() {
    return getLong(HISTORY_DEFAULTPERIOD, 1000);
  }

  /**
   * Maximum number of discrete history events to hold onto for reporting.
   */
  public long getHistoryMaxLength() {
    return getLong(HISTORY_MAXLENGTH, 300);
  }

  /**
   * Polling period in millis for tail to check a file.
   * 
   * Currently we poll to see if a file has changed and this allows the user to
   * tweak the frequency
   */
  public long getTailPollPeriod() {
    return getLong(TAIL_POLLPERIOD, 1000);
  }

  public String getCollectorHost() {
    return get(COLLECTOR_EVENT_HOST, "localhost");
  }

  public int getCollectorPort() {
    // FLUME = 35863 (flume on the telephone)
    return getInt(COLLECTOR_EVENT_PORT, 35853);
  }

  /**
   * This returns the type of RPC mechanism (Thrift or Avro) chosen for the
   * FlumeEventServer.
   */
  public String getEventRPC() {
    String[] validRPCProtocols = { RPC_TYPE_AVRO, RPC_TYPE_THRIFT };
    String entered = get(EVENT_RPC_TYPE, RPC_TYPE_THRIFT).toUpperCase();
    for (String prot : validRPCProtocols) {
      if (entered.equals(prot)) {
        return prot;
      }
    }
    // defaulting to Thrift with a polite warning
    LOG.warn("flume.event.rpc = '{}' incorrectly defined, should be either"
        + " \"THRIFT\" or \"AVRO\".  Defaulting to \"THRIFT\"", entered);
    return RPC_TYPE_THRIFT;
  }

  public String getCollectorDfsDir() {
    return get(COLLECTOR_DFS_DIR, "file://tmp/flume-${user.name}/collected");
  }

  public String getCollectorDfsCompressCodec() {
    return get(COLLECTOR_DFS_COMPRESS_CODEC, "None");
  }

  public long getCollectorRollMillis() {
    return getLong(COLLECTOR_ROLL_MILLIS, 30000);
  }

  /**
   * This is the list of masters that agent nodes will connect to
   */
  public String getMasterServers() {
    String svrs = get(MASTER_SERVERS, "localhost");

    // check for illegal ':'s in the servers; truncate stuff after the ':'
    String[] hosts = svrs.split(",");
    StringBuilder builder = new StringBuilder();
    for (int i = 0; i < hosts.length; ++i) {
      hosts[i] = hosts[i].trim();

      String[] parts = hosts[i].split(":");
      builder.append(parts[0]);
      if (parts.length > 1) {
        LOG.warn("Master Server's should not have list ports but host '"
            + hosts[i] + " 'specified ports! ");
      }

      if (i < hosts.length - 1) {
        builder.append(",");
      }
    }

    return builder.toString();
  }

  /**
   * Return a list of server:port pairs for heartbeat purposes
   */
  public String getMasterHeartbeatServers() {
    String servers = get(MASTER_HEARTBEAT_SERVERS, null);
    if (servers != null) {
      return servers;
    }

    String[] hosts = getMasterServers().split(",");
    int heartbeatport = getMasterHeartbeatPort();
    List<String> l = Arrays.asList(hosts);
    Iterator<String> iter = l.iterator();
    StringBuilder builder = new StringBuilder();
    while (iter.hasNext()) {
      builder.append(iter.next() + ":" + heartbeatport);
      if (iter.hasNext()) {
        builder.append(',');
      }
    }
    return builder.toString();
  }

  /**
   * Utility function to retrieve master hosts as a list of host / port pairs
   * used only on the client!
   */
  public List<Pair<String, Integer>> getMasterHeartbeatServersList() {
    String[] hosts = getMasterHeartbeatServers().split(",");
    List<Pair<String, Integer>> ret = new ArrayList<Pair<String, Integer>>(
        hosts.length);
    for (String s : hosts) {
      String[] parts = s.split(":");
      ret.add(new Pair<String, Integer>(parts[0], Integer.parseInt(parts[1])));
    }
    return ret;
  }

  public int getMasterHttpPort() {
    return getInt(MASTER_HTTP_PORT, DEFAULT_HTTP_PORT);
  }

  /**
   * The port on which to start a ReportServer
   */
  public int getReportServerPort() {
    return getInt(REPORT_SERVER_PORT, DEFAULT_REPORT_SERVER_PORT);
  }

  /**
   * This returns the type of RPC mechanism (Thrift or Avro) chosen for the
   * FlumeReportServer.
   */
  public String getReportServerRPC() {
    String[] validRPCProtocols = { RPC_TYPE_AVRO, RPC_TYPE_THRIFT };
    String entered = get(REPORT_SERVER_RPC_TYPE, RPC_TYPE_THRIFT).toUpperCase();
    for (String prot : validRPCProtocols) {
      if (entered.equals(prot)) {
        return prot;
      }
    }
    // defaulting to Thrift with a polite warning
    LOG.warn("flume.report.server.rpc.type incorrectly defined, "
        + "should be either \"THRIFT\" or \"AVRO\".  "
        + "Defaulting to \"THRIFT\"");
    return RPC_TYPE_THRIFT;
  }

  /**
   * If MASTER_HEARTBEAT_PORT is set, we use that as our heartbeat port. If not,
   * we look at the list of server:port pairs in MASTER_HEARTBEAT_SERVERS, in
   * particular we look at the MASTER_SERVER_ID'th value
   */
  public int getMasterHeartbeatPort() {
    String port = get(MASTER_HEARTBEAT_PORT, null);
    if (port != null) {
      return Integer.parseInt(port);
    }
    String servers = get(MASTER_HEARTBEAT_SERVERS, null);
    if (servers == null) {
      return DEFAULT_HEARTBEAT_PORT;
    }

    // MASTER_HEARTBEAT_SERVERS is set - split it and guess at our server port
    List<String> serverList = Arrays.asList(servers.split(","));
    int serverid = getMasterServerId();
    Preconditions.checkState(serverid < serverList.size(),
        "Serverid is out of range: " + serverid);

    String[] server = serverList.get(serverid).split(":");
    Preconditions.checkState(server.length == 2, "Server spec "
        + serverList.get(serverid) + " is ill-formed");
    return Integer.parseInt(server[1].trim());
  }

  public String getMasterHeartbeatRPC() {
    String[] validRPCProtocols = { RPC_TYPE_AVRO, RPC_TYPE_THRIFT };
    String entered = get(MASTER_HEARBEAT_RPC, RPC_TYPE_THRIFT).toUpperCase();
    for (String prot : validRPCProtocols) {
      if (entered.equals(prot)) {
        return prot;
      }
    }
    // default
    return RPC_TYPE_THRIFT;
  }

  public int getMasterGossipPeriodMs() {
    return getInt(MASTER_GOSSIP_PERIOD_MS, 1000);
  }

  /**
   * Max age for gossiped acks to stick around.
   */
  public long getMasterGossipMaxAgeMs() {
    return getLong(MASTER_GOSSIP_MAXAGE_MS, 300 * 1000);
  }

  public int getMasterGossipPort() {
    String port = get(MASTER_GOSSIP_PORT, null);
    if (port != null) {
      return Integer.parseInt(port);
    }
    // gossipport is not set, try and guess from MASTER_GOSSIP_SERVERS if it's
    // set
    // We can't call getMasterGossipServers because that might call into this
    // method
    // itself...
    String servers = get(MASTER_GOSSIP_SERVERS, null);
    if (servers == null) {
      return DEFAULT_GOSSIP_PORT;
    }

    // MASTER_GOSSIP_SERVERS is set - split it and guess at our server port
    List<String> serverList = Arrays.asList(servers.split(","));
    int serverid = getMasterServerId();
    Preconditions.checkState(serverid < serverList.size(),
        "Serverid is out of range: " + serverid);

    String[] server = serverList.get(serverid).split(":");
    Preconditions.checkState(server.length == 2, "Server spec "
        + serverList.get(serverid) + " is ill-formed");
    return Integer.parseInt(server[1].trim());
  }

  /**
   * Returns a comma-separated list of host:port pairs. If not explicitly set,
   * defaults to MASTER_SERVERS, using MASTER_GOSSIP_PORT.
   */
  public String getMasterGossipServers() {
    String ret = get(MASTER_GOSSIP_SERVERS, null);
    if (ret != null) {
      return ret;
    }

    // Otherwise return the list of ensemble hosts and the gossip port
    String[] hosts = getMasterServers().split(",");
    List<String> l = Arrays.asList(hosts);
    Iterator<String> iter = l.iterator();
    StringBuilder builder = new StringBuilder();
    while (iter.hasNext()) {
      builder.append(iter.next() + ":" + getMasterGossipPort());
      if (iter.hasNext()) {
        builder.append(',');
      }
    }
    return builder.toString();
  }

  /**
   * Utility function to return a list of host:port pairs corresponding to all
   * the gossip hosts in a Flume Master ensemble.
   */
  public List<Pair<String, Integer>> getConfigMasterGossipHostsList() {
    String hostString = getMasterGossipServers();
    if (hostString.equals("")) {
      return new ArrayList<Pair<String, Integer>>();
    }
    String[] hosts = hostString.split(",");
    List<Pair<String, Integer>> ret = new ArrayList<Pair<String, Integer>>(
        hosts.length);
    for (String s : hosts) {
      String[] parts = s.split(":");
      ret.add(new Pair<String, Integer>(parts[0], Integer.parseInt(parts[1])));
    }
    return ret;
  }

  public long getConfigHeartbeatPeriod() {
    return getLong(CONFIG_HEARTBEAT_PERIOD, 5000); // millis
  }

  public int getMasterMaxMissedheartbeats() {
    return getInt(MASTER_HEARTBEAT_MAX_MISSED, 10);
  }

  public long getNodeHeartbeatBackoffLimit() {
    return getLong(NODE_HEARTBEAT_BACKOFF_LIMIT, 60000); // 1 minute in millis
  }

  public boolean getNodeAutofindHttpPort() {
    return getBoolean(NODE_HTTP_AUTOFINDPORT, true);
  }

  public int getConfigAdminPort() {
    return getInt(CONFIG_ADMIN_PORT, DEFAULT_ADMIN_PORT);
  }

  public int getConfigReportPort() {
    return getInt(REPORT_SERVER_PORT, DEFAULT_REPORT_SERVER_PORT);
  }

  public String getMasterSavefile() {
    return get(MASTER_SAVEFILE, "conf/current.flume");
  }

  public boolean getMasterSavefileAutoload() {
    return getBoolean(MASTER_SAVEFILE_AUTOLOAD, false);
  }

  public String getKerberosPrincipal() {
    // flume is the default kerberos principal for hdfs writes
    // _HOST is expanded by the hadoop security util to be the local host name.
    return get(SECURITY_KERBEROS_PRINCIPAL, "");
  }

  public String getKerberosKeytab() {
    // mit kerb v5 default keytab path
    return get(SECURITY_KERBEROS_KEYTAB, "/etc/flume/conf/krb5.keytab");
  }

  public long getReporterPollPeriod() {
    return getLong(REPORTER_POLLER_PERIOD, 2000); // millis
  }

  public String getFlurkerEncoding() {
    return get(FLURKER_ENCODING, "UTF-8");
  }

  public String getWebAppsPath() {
    return get(WEBAPPS_PATH, "webapps");
  }

  /**
   * This method loads the configuration, or does a hard exit if loading the
   * configuration fails.
   */
  public static FlumeConfiguration hardExitLoadConfig() {

    try {
      FlumeConfiguration conf = FlumeConfiguration.get();

      // Loading the config is lazy! need to try to get something before an
      // exception is thrown.

      conf.getMaxRestartsPerMin(); // get some random value.
      return conf;
    } catch (RuntimeException parseEx) {
      System.exit(-1);
    }
    return null; // dead code
  }

  public String getTwitterName() {
    return get(TWITTER_USER, "");
  }

  public String getTwitterPW() {
    return get(TWITTER_PW, "");
  }

  public String getTwitterURL() {
    return get(TWITTER_STREAM_URL,
        "http://stream.twitter.com/1/statuses/sample.json");
  }

  public String getDefaultOutputFormat() {
    // Look at FormatFactory for possible values.
    return get(COLLECTOR_OUTPUT_FORMAT, "avrojson");
  }

  public String getGangliaServers() {
    // gmond's default multicast ip and port
    return get(GANGLIA_SERVERS, "239.2.11.71:8649");
  }

  public long getEventMaxSizeBytes() {
    return getLong(EVENT_MAX_SIZE, 32 * 1024); // Default to 32k
  }

  public String getPluginClasses() {
    return get(PLUGIN_CLASSES, "");
  }

  public String getHiveHost() {
    return get(HIVE_HOST, "localhost");
  }

  public int getHivePort() {
    return getInt(HIVE_PORT, 9083); // default port found from meta store
    // presentation.
  }

  public String getHiveUser() {
    return get(HIVE_USER, "hive");
  }

  public String getHiveUserPW() {
    return get(HIVE_PW, "");
  }

  public long getHeartbeatBackoff() {
    return getLong(AGENT_HEARTBEAT_BACKOFF, 1000); // millis
  }

  /**
   * Max milliseconds to back off after a close request. Close will wait at most
   * this amount of time after it sees progress after a close request. * after a
   * close request.
   */
  public long getThriftCloseMaxSleep() {
    return getLong(THRIFT_CLOSE_MAX_SLEEP, 10 * 1000);
  }

  /**
   * The default port for scribe sources to listen on
   */
  public int getScribeSourcePort() {
    return getInt(SCRIBE_SOURCE_PORT, DEFAULT_SCRIBE_SOURCE_PORT);
  }

  /**
   * Returns the default flow name for logical nodes.
   */
  public String getDefaultFlowName() {
    return get(DEFAULT_FLOW_NAME, "default-flow");
  }

  /**
   * Returns the webapp root for the master.
   */
  public String getMasterWebappRoot() {
    String home = getFlumeHome();
    if (home == null) {
      home = ".";
    }
    return home + File.separator + get(WEBAPP_ROOT_MASTER,
      "webapps/flumemaster.war");
  }

  /**
   * Returns the webapp root for the node (i.e. agent or collector).
   */
  public String getNodeWebappRoot() {
    String home = getFlumeHome();
    if (home == null) {
      home = ".";
    }
    return home + File.separator + get(WEBAPP_ROOT_NODE,
        "webapps/flumeagent.war");
  }

  /**
   * Returns the current FlumeConfiguration as an HTML string
   */
  public String toHtml() {
    Iterator<Entry<String, String>> iter = iterator();
    ArrayList<String> keys = new ArrayList<String>();
    while (iter.hasNext()) {
      Entry<String, String> e = iter.next();
      keys.add(e.getKey());
    }
    Collections.sort(keys);

    StringWriter sw = new StringWriter();
    PrintWriter pw = new PrintWriter(sw);

    pw.print("<table>");

    for (String k : keys) {
      String value = get(k);
      pw.println("<tr><th>" + k + "</th>");
      pw.print("<td>");
      pw.print("<div class=\"" + k + "\">");
      pw.print(value);
      pw.print("</div>");
      pw.println("</td>");
      pw.println("</tr>");
    }

    pw.print("</table>");

    pw.flush();

    return sw.getBuffer().toString();
  }

  /**
   * If a logical node does not cleanly close after the specified amount of
   * time, it is is interrupted and should exit in error state. This can occur
   * because of some blocking sources, sinks, or decorators.
   */
  public long getNodeCloseTimeout() {
    return getLong(NODE_CLOSE_TIMEOUT, 30000);
  }

  /**
   * Whether or not the WAL should buffer writes to disk. Defaults to true.
   */
  public boolean getWALOutputBuffering() {
    return getBoolean(WAL_OUTPUT_BUFFER, true);
  }

} public class SinkFactoryImpl extends SinkFactory {
  static final Logger LOG = LoggerFactory.getLogger(SinkFactoryImpl.class);

  // The actual types are <String, SinkBuilder>
  static Object[][] sinkList = {
      // high level sinks.
      { "collectorSink", CollectorSink.hdfsBuilder() },

      { "agentSink", AgentSink.e2eBuilder() },
      { "agentE2ESink", AgentSink.e2eBuilder() }, // now with acks
      { "agentDFOSink", AgentSink.dfoBuilder() },
      { "agentBESink", AgentSink.beBuilder() },

      { "agentFailoverSink", AgentSink.dfoBuilder() },
      { "agentBestEffortSink", AgentSink.beBuilder() },

      { "agentE2EChain", AgentFailChainSink.e2eBuilder() },
      { "agentDFOChain", AgentFailChainSink.dfoBuilder() },
      { "agentBEChain", AgentFailChainSink.beBuilder() },

      // autoE2EChain, autoDFOChain and autoBEChains are essentially node
      // specific "macros", and use let expression shadowing
      { "autoBEChain", EventSink.StubSink.builder("autoBEChain") },
      { "autoDFOChain", EventSink.StubSink.builder("autoDFOChain") },
      { "autoE2EChain", EventSink.StubSink.builder("autoE2EChain") },
      { "logicalSink", EventSink.StubSink.builder("logicalSink") },

      // low level sinks
      { "null", NullSink.builder() },
      // all calls throw exception
      { "fail", EventSink.StubSink.builder("fail") },

      { "console", ConsoleEventSink.builder() },
      { "text", TextFileSink.builder() },
      { "seqfile", SeqfileEventSink.builder() },
      { "dfs", DFSEventSink.builder() }, // escapes
      { "formatDfs", CustomDfsSink.builder() }, // does not escape
      { "escapedFormatDfs", EscapedCustomDfsSink.builder() }, // escapes
      { "customdfs", CustomDfsSink.builder() }, // TODO deprecate
      { "escapedCustomDfs", EscapedCustomDfsSink.builder() }, // TODO deprecate

      { "rpcSink", RpcSink.builder() }, // creates AvroEventSink or
      // ThriftEventSink
      { "syslogTcp", SyslogTcpSink.builder() },
      { "irc", IrcSink.builder() },
      { "thriftSink", ThriftEventSink.builder() },
      { "avroSink", AvroEventSink.builder() },
      // advanced
      { "failChain", FailoverChainSink.builder() }, // @deprecated

      // reports
      { "ganglia", GangliaSink.builder() },
      { "counter", CounterSink.builder() },
      { "accumulator", AccumulatorSink.builder() },
      { "counterHistory", CountHistoryReporter.builder() },
      { "multigrep", MultiGrepReporterSink.builderSimple() },
      { "multigrepspec", MultiGrepReporterSink.builder() },
      { "regexhisto", RegexGroupHistogramSink.builderSimple() },
      { "regexhistospec", RegexGroupHistogramSink.builder() },

  };

  // The actual types are <String, SinkDecoBuilder>
  static Object[][] decoList = {
      { "nullDeco", EventSinkDecorator.nullBuilder() },
      { "diskFailover", DiskFailoverDeco.builder() },
      { "ackedWriteAhead", NaiveFileWALDeco.builderEndToEndDir() },

      { "ackChecker", AckChecksumChecker.builder() },
      { "ackInjector", AckChecksumInjector.builder() },

      { "lazyOpen", LazyOpenDecorator.builder() },
      { "insistentOpen", InsistentOpenDecorator.builder() },
      { "insistentAppend", InsistentAppendDecorator.builder() },
      { "stubbornAppend", StubbornAppendSink.builder() },

      // relational algebra projection
      { "value", ValueDecorator.builder() },
      { "mask", MaskDecorator.builder() },
      { "select", SelectDecorator.builder() },

      // TODO (jon) relational algebra selection (filtering)

      // format the output
      { "format", FormatterDecorator.builder() },

      // message digest of body
      { "digest", DigestDecorator.builder() },

      // extractors
      { "regex", RegexExtractor.builder() },
      { "regexAll", RegexAllExtractor.builder() },
      { "split", SplitExtractor.builder() },
      { "exDate", DateExtractor.builder() },

      // cpu / network tradeoffs
      { "batch", BatchingDecorator.builder() },
      { "unbatch", UnbatchingDecorator.builder() },
      { "gzip", GzipDecorator.builder() },
      { "gunzip", GunzipDecorator.builder() },

      // sampling
      { "intervalSampler", IntervalSampler.builder() },
      { "probSampler", ProbabilitySampler.builder() },
      { "reservoirSampler", ReservoirSamplerDeco.builder() },

      // debugging related
      { "flakeyAppend", FlakeyEventSink.builder() },
      { "intervalFlakeyAppend", IntervalFlakeyEventSink.builder() },
      { "intervalDroppyAppend", IntervalDroppyEventSink.builder() },

      { "inmem", InMemoryDecorator.builder() },
      { "benchinject", BenchmarkInjectDecorator.builder() },
      { "benchreport", BenchmarkReportDecorator.builder() },
      { "bloomGen", BloomGeneratorDeco.builder() },
      { "bloomCheck", BloomCheckDecorator.builder() },
      { "mult", MultiplierDecorator.builder() },
      { "delay", DelayDecorator.builder() },
      { "choke", ChokeDecorator.builder() },

  };

  Map<String, SinkBuilder> sinks = new HashMap<String, SinkBuilder>();
  Map<String, SinkDecoBuilder> decos = new HashMap<String, SinkDecoBuilder>();

  public SinkFactoryImpl() {
    for (Object[] entry : sinkList) {
      String key = (String) entry[0];
      SinkBuilder value = (SinkBuilder) entry[1];
      sinks.put(key, value);
    }

    for (Object[] entry : decoList) {
      String key = (String) entry[0];
      SinkDecoBuilder value = (SinkDecoBuilder) entry[1];
      decos.put(key, value);
    }

    String classes = FlumeConfiguration.get().getPluginClasses();
    if (!classes.equals("")) {
      for (String s : classes.split(",")) {
        loadPluginBuilders(s);
      }
    }
  }

  @Override
  public Set<String> getSinkNames() {
    return Collections.unmodifiableSet(sinks.keySet());
  }

  @Override
  public Set<String> getDecoratorNames() {
    return Collections.unmodifiableSet(decos.keySet());
  }

  /**
   * Given a fully qualified classname, load the class it represents and try and
   * invoke two methods to get a list of sinks and builders.
   * 
   * This should only be called at startup to avoid dependency issues; there is
   * no lazy loading.
   * 
   * Any exceptions are simply logged and ignored. This is an undocumented
   * feature for now.
   */
  @SuppressWarnings("unchecked")
  protected void loadPluginBuilders(String clsName) {
    try {
      Class<Object> cls = (Class<Object>) Class.forName(clsName);
      try {
        Method mth = cls.getMethod("getSinkBuilders");
        List<Pair<String, SinkBuilder>> builders = (List<Pair<String, SinkBuilder>>) mth
            .invoke(mth);
        for (Pair<String, SinkBuilder> snk : builders) {
          LOG.info("Found sink builder " + snk.getLeft() + " in " + clsName);
          sinks.put(snk.getLeft(), snk.getRight());
        }
      } catch (NoSuchMethodException e) {
        LOG.warn("No sink builders found in " + clsName);
      } catch (Exception e) {
        LOG.error("Error invoking getSinkBuilders on class " + clsName, e);
      }
      try {
        Method mth = cls.getMethod("getDecoratorBuilders");
        List<Pair<String, SinkDecoBuilder>> builders = (List<Pair<String, SinkDecoBuilder>>) mth
            .invoke(mth);
        for (Pair<String, SinkDecoBuilder> snk : builders) {
          LOG.info("Found sink decorator " + snk.getLeft() + " in " + clsName);
          decos.put(snk.getLeft(), snk.getRight());
        }
      } catch (NoSuchMethodException e) {
        LOG.warn("No sink decorators found in " + clsName);
      } catch (Exception e) {
        LOG.error("Error invoking getDecoratorBuilders on class " + clsName, e);
      }
    } catch (ClassNotFoundException e) {
      LOG.error("Could not find class " + clsName + " for plugin loading", e);
    }
  }

  @Override
  public EventSinkDecorator<EventSink> getDecorator(Context context,
      String name, String... args) throws FlumeSpecException {
    try {
      SinkDecoBuilder builder = decos.get(name);
      if (builder == null)
        return null;
      return builder.build(context, args);
    } catch (NumberFormatException nfe) {
      throw new FlumeArgException("Illegal number format: " + nfe.getMessage());
    } catch (IllegalArgumentException iae) {
      throw new FlumeArgException(iae.getMessage());
    }
  }

  @Override
  public EventSink createSink(Context context, String name, Object... args)
      throws FlumeSpecException {
    try {
      SinkBuilder builder = sinks.get(name);
      if (builder == null)
        return null;
      return builder.create(context, args);
    } catch (NumberFormatException nfe) {
      throw new FlumeArgException("Illegal number format: " + nfe.getMessage());
    } catch (IllegalArgumentException iae) {
      throw new FlumeArgException(iae.getMessage());
    }
  }

  @Override
  public EventSink getSink(Context context, String name, String... args)
      throws FlumeSpecException {
    try {
      SinkBuilder builder = sinks.get(name);
      if (builder == null)
        return null;
      return builder.build(context, args);
    } catch (NumberFormatException nfe) {
      throw new FlumeArgException("Illegal number format: " + nfe.getMessage());
    } catch (IllegalArgumentException iae) {
      throw new FlumeArgException(iae.getMessage());
    }
  }

  /**
   * This is only for testing
   */
  public void setSink(String name, SinkBuilder builder) {
    sinks.put(name, builder);
  }

  /**
   * This is only for testing
   */
  public void setDeco(String name, SinkDecoBuilder builder) {
    decos.put(name, builder);
  }
} abstract public class Event {
  static final Logger LOG = LoggerFactory.getLogger(Event.class);

  public final static String A_SERVICE = "service";

  public enum Priority {
    FATAL, ERROR, WARN, INFO, DEBUG, TRACE
  };

  /**
   * the body - a blob of raw bytes that contains the raw entry. Values can be
   * extracted from this body but must not change the body. This should never
   * return null. To change an even body, one should create a new event.
   */
  abstract public byte[] getBody();

  /**
   * the priority - user specified priority
   */
  abstract public Priority getPriority();

  /**
   * a time stamp - unix millis
   */
  abstract public long getTimestamp();

  /**
   * a nano time - for ordering if entries have the same millis
   */
  abstract public long getNanos();

  /**
   * Host name of the machine that generated this event.
   * 
   * TODO (jon) consider wrapping this. Chose string because it doesn't assume
   * ipv4 or ipv6, etc. May cause canonalicalization problems.
   */
  abstract public String getHost();

  /**
   * This gets a particular attribute added to an event. This an extensible
   * interface for "other" attributes.
   */
  abstract public byte[] get(String attr);

  /**
   * This sets a particular attribute to an event. Attributes are write once,
   * and overwrites of an attribute are not permitted. If a call attempts to
   * overwrite an attribute, a IllegalArgumentException will be thrown.
   * 
   * This method copies the reference, and does not copy the data. (shallow
   * copy, not deep)
   */
  abstract public void set(String attr, byte[] v)
      throws IllegalArgumentException;

  /**
   * Ideally this would be package private, but instead this method should
   * return a ReadOnly Map. This can be done by wrapping a map m with
   * Collections.unmodifiableMap(m).
   */
  abstract public Map<String, byte[]> getAttrs();

  /**
   * Constructs the union of this event and another. The body in this event is
   * preserved, as are any attributes that already exist in this event.
   */
  abstract public void merge(Event e);

  /**
   * Constructs the union of this event and another. The body of this event is
   * preserved, but all attributes in e are renamed with the supplied prefix to
   * try to avoid name collisions. In the event of a collision, attributes from
   * this event are preserved.
   */
  abstract public void hierarchicalMerge(String prefix, Event e);

  /**
   * These are useful to other classes which might want to search for tags in
   * strings.
   */
  final public static String TAG_REGEX = "\\%(\\w|\\%)|\\%\\{([\\w\\.-]+)\\}";
  final public static Pattern tagPattern = Pattern.compile(TAG_REGEX);

  /**
   * Returns true if in contains a substring matching TAG_REGEX (i.e. of the
   * form %{...} or %x.
   */
  public static boolean containsTag(String in) {
    return tagPattern.matcher(in).find();
  }

  protected String expandShorthand(char c) {
    // It's a date
    switch (c) {
    case 'a':
      return "weekday_short";
    case 'A':
      return "weekday_full";
    case 'b':
      return "monthname_short";
    case 'B':
      return "monthname_full";
    case 'c':
      return "datetime";
    case 'd':
      return "day_of_month_xx"; // two digit
    case 'D':
      return "date_short"; // "MM/dd/yy";
    case 'H':
      return "hour_24_xx";
    case 'I':
      return "hour_12_xx";
    case 'j':
      return "day_of_year_xxx"; // three digits
    case 'k':
      return "hour_24"; // 1 or 2 digits
    case 'l':
      return "hour_12"; // 1 or 2 digits
    case 'm':
      return "month_xx";
    case 'M':
      return "minute_xx";
    case 'p':
      return "am_pm";
    case 's':
      return "unix_seconds";
    case 'S':
      return "seconds_xx";
    case 't':
      // This is different from unix date (which would insert a tab character
      // here)
      return "unix_millis";
    case 'y':
      return "year_xx";
    case 'Y':
      return "year_xxxx";
    case 'z':
      return "timezone_delta";
    default:
      LOG.warn("Unrecognized escape in event format string: %" + c);
      return "" + c;
    }

  }

  /**
   * Hardcoded lookups for %x style escape replacement. Add your own!
   * 
   * All shorthands are Date format strings, currently.
   * 
   * Returns the empty string if an escape is not recognized.
   * 
   * Dates follow the same format as unix date, with a few exceptions.
   * 
   */
  protected String replaceShorthand(char c) {
    // It's a date
    String formatString = "";
    switch (c) {
    case '%':
      return "%";
    case 'a':
      formatString = "EEE";
      break;
    case 'A':
      formatString = "EEEE";
      break;
    case 'b':
      formatString = "MMM";
      break;
    case 'B':
      formatString = "MMMM";
      break;
    case 'c':
      formatString = "EEE MMM d HH:mm:ss yyyy";
      break;
    case 'd':
      formatString = "dd";
      break;
    case 'D':
      formatString = "MM/dd/yy";
      break;
    case 'H':
      formatString = "HH";
      break;
    case 'I':
      formatString = "hh";
      break;
    case 'j':
      formatString = "DDD";
      break;
    case 'k':
      formatString = "H";
      break;
    case 'l':
      formatString = "h";
      break;
    case 'm':
      formatString = "MM";
      break;
    case 'M':
      formatString = "mm";
      break;
    case 'p':
      formatString = "a";
      break;
    case 's':
      return "" + getTimestamp() / 1000;
    case 'S':
      formatString = "ss";
      break;
    case 't':
      // This is different from unix date (which would insert a tab character
      // here)
      return Long.valueOf(getTimestamp()).toString();
    case 'y':
      formatString = "yy";
      break;
    case 'Y':
      formatString = "yyyy";
      break;
    case 'z':
      formatString = "ZZZ";
      break;
    default:
      LOG.warn("Unrecognized escape in event format string: %" + c);
      return "";
    }
    SimpleDateFormat format = new SimpleDateFormat(formatString);
    Date date = new Date(getTimestamp());
    return format.format(date);
  }

  /**
   * Looks up an attribute for a given tag name, but checks for some special
   * cases first. Used for string escaping.
   */
  protected String mapTagToString(String tag) {
    if (tag.equals("hostname") || tag.equals("host")) {
      return getHost();
    }
    if (tag.equals("nanos")) {
      return "" + getNanos();
    }
    if (tag.equals("priority")) {
      return getPriority().toString();
    }
    if (tag.equals("body")) {
      return new String(getBody());
    }

    byte[] attr = get(tag);
    if (attr != null) {
      return new String(attr);
    }
    return null;
  }

  /**
   * Replace all substrings of form %{tagname} with get(tagname).toString() and
   * all shorthand substrings of form %x with a special value.
   * 
   * Any unrecognized / not found tags will be replaced with the empty string.
   * 
   * TODO(henry): we may want to consider taking this out of Event and into a
   * more general class when we get more use cases for this pattern.
   */
  public String escapeString(String in) {
    Matcher matcher = tagPattern.matcher(in);
    StringBuffer sb = new StringBuffer();
    while (matcher.find()) {
      String replacement = "";
      // Group 2 is the %{...} pattern
      if (matcher.group(2) != null) {

        replacement = mapTagToString(matcher.group(2));
        if (replacement == null) {
          replacement = "";
          LOG.warn("Tag " + matcher.group(2) + " not found");
        }
      } else {
        // The %x pattern.
        // Since we know the match is a single character, we can
        // switch on that rather than the string.
        Preconditions.checkState(matcher.group(1) != null
            && matcher.group(1).length() == 1,
            "Expected to match single character tag in string " + in);
        replacement = replaceShorthand(matcher.group(1).charAt(0));
      }

      // The replacement string must have '$' and '\' chars escaped. This
      // replacement string is pretty arcane.
      //
      // replacee : '$' -> for java '\$' -> for regex "\\$"
      // replacement: '\$' -> for regex '\\\$' -> for java "\\\\\\$"
      //
      // replacee : '\' -> for java "\\" -> for regex "\\\\"
      // replacement: '\\' -> for regex "\\\\" -> for java "\\\\\\\\"

      // note: order matters
      replacement = replacement.replaceAll("\\\\", "\\\\\\\\");
      replacement = replacement.replaceAll("\\$", "\\\\\\$");

      matcher.appendReplacement(sb, replacement);
    }
    matcher.appendTail(sb);
    return sb.toString();
  }

  /**
   * Instead of replacing escape sequences in a string, this method returns a
   * mapping of an attribute name to the value based on the escape sequence
   * found in the argument string.
   */
  public Map<String, String> getEscapeMapping(String in) {
    Map<String, String> mapping = new HashMap<String, String>();
    Matcher matcher = tagPattern.matcher(in);
    while (matcher.find()) {
      String replacement = "";
      // Group 2 is the %{...} pattern
      if (matcher.group(2) != null) {

        replacement = mapTagToString(matcher.group(2));
        if (replacement == null) {
          replacement = "";
          LOG.warn("Tag " + matcher.group(2) + " not found");
        }
        mapping.put(matcher.group(2), replacement);
      } else {
        // The %x pattern.
        // Since we know the match is a single character, we can
        // switch on that rather than the string.
        Preconditions.checkState(matcher.group(1) != null
            && matcher.group(1).length() == 1,
            "Expected to match single character tag in string " + in);
        char c = matcher.group(1).charAt(0);
        replacement = replaceShorthand(c);
        mapping.put(expandShorthand(c), replacement);
      }
    }
    return mapping;

  }
} public class AckChecksumChecker<S extends EventSink> extends
    EventSinkDecorator<S> {
  static final Logger LOG = LoggerFactory.getLogger(AckChecksumChecker.class);

  final static public String A_ACK_STARTS = "ackStarts";
  final static public String A_ACK_ENDS = "ackEnds";
  final static public String A_ACK_FAILS = "ackFails";
  final static public String A_ACK_SUCCESS = "ackSuccesses";
  final static public String A_ACK_UNEXPECTED = "ackUnexpected";

  AtomicLong ackStarts = new AtomicLong();
  AtomicLong ackEnds = new AtomicLong();
  AtomicLong ackFails = new AtomicLong();
  AtomicLong ackSuccesses = new AtomicLong();

  // TODO (jon) this is very inefficient right now.
  Map<String, Long> partial = new HashMap<String, Long>();

  final AckListener listener;
  long unstarted = 0; // number of events that didn't have a start event.

  public AckChecksumChecker(S s, AckListener l) {
    super(s);
    Preconditions.checkNotNull(l);
    this.listener = l;
  }

  public AckChecksumChecker(S s) {
    super(s);
    // do nothing listener
    this.listener = new AckListener() {
      @Override
      public void end(String group) {
        LOG.info("ended " + group);
      }

      @Override
      public void err(String group) {
        LOG.info("erred " + group);

      }

      @Override
      public void start(String group) {
        LOG.info("start " + group);

      }

      @Override
      public void expired(String group) throws IOException {
        LOG.info("expired " + group);
      }
    };
  }

  @Override
  public void append(Event e) throws IOException, InterruptedException {
    byte[] btyp = e.get(AckChecksumInjector.ATTR_ACK_TYPE);

    if (btyp == null) {
      // pass through if has no checksumming tags
      super.append(e);
      return;
    }

    byte[] btag = e.get(AckChecksumInjector.ATTR_ACK_TAG);
    byte[] bchk = e.get(AckChecksumInjector.ATTR_ACK_HASH);
    String k = new String(btag);

    if (Arrays.equals(btyp, AckChecksumInjector.CHECKSUM_START)) {
      LOG.info("Starting checksum group called " + k);
      // Checksum Start marker: create new partial
      long newchk = ByteBuffer.wrap(bchk).getLong();
      LOG.info("initial checksum is " + Long.toHexString(newchk));
      partial.put(k, newchk);
      ackStarts.incrementAndGet();
      listener.start(k);

      return;
    } else if (Arrays.equals(btyp, AckChecksumInjector.CHECKSUM_STOP)) {
      LOG.info("Finishing checksum group called '" + k + "'");
      ackEnds.incrementAndGet();
      // Checksum stop marker: move from partial to done
      Long chksum = partial.get(k);
      if (chksum == null) {
        LOG.error("checksum failed");
        listener.err(k);
        ackFails.incrementAndGet();
        return;
      }
      long endchk = ByteBuffer.wrap(bchk).getLong();
      LOG.debug("final checksum is " + Long.toHexString(endchk)
          + " stop checksum is " + Long.toHexString(chksum));
      if ((chksum ^ endchk) != 0) {
        // There was a problem.

        LOG.warn("[ Thread " + Thread.currentThread().getId()
            + " ] Some component of msg group was lost or duped " + k);
        listener.err(k);
        ackFails.incrementAndGet();
        return;
      }

      LOG.info("Checksum succeeded " + Long.toHexString(chksum));
      listener.end(k);
      ackSuccesses.incrementAndGet();
      partial.remove(k);
      LOG.info("moved from partial to complete " + k);
      return;
    }

    // normal case, just an update the checksum.
    CRC32 chk = new CRC32();
    chk.reset();
    chk.update(e.getBody());
    long chkVal = chk.getValue();

    if (chkVal != ByteBuffer.wrap(bchk).getLong()) {
      LOG.warn("check sum does not match!");
    }
    super.append(e);

    // only do this after we have successfully sent the event.
    synchronized (partial) {
      Long chks = partial.get(k);
      if (chks == null) {
        // throw new IOException("Ack tag '" + k + "' was not started: ");
        unstarted++;
        return;
      }

      // update checksum.
      long checksum = partial.get(k);
      checksum ^= chkVal;
      partial.put(k, checksum);
    }
  }

  @Override
  public void close() throws IOException, InterruptedException {
    super.close();
    if (partial.size() != 0) {
      LOG.warn("partial acks abandoned: " + partial);
    }
  }

  @Override
  public ReportEvent getMetrics() {
    ReportEvent rpt = super.getMetrics();
    rpt.setLongMetric(A_ACK_FAILS, ackFails.get());
    rpt.setLongMetric(A_ACK_SUCCESS, ackSuccesses.get());
    rpt.setLongMetric(A_ACK_STARTS, ackStarts.get());
    rpt.setLongMetric(A_ACK_ENDS, ackEnds.get());
    rpt.setLongMetric(A_ACK_UNEXPECTED, unstarted);
    return rpt;
  }

  @Deprecated
  @Override
  public ReportEvent getReport() {
    ReportEvent rpt = super.getReport();
    rpt.setLongMetric(A_ACK_FAILS, ackFails.get());
    rpt.setLongMetric(A_ACK_SUCCESS, ackSuccesses.get());
    rpt.setLongMetric(A_ACK_STARTS, ackStarts.get());
    rpt.setLongMetric(A_ACK_ENDS, ackEnds.get());
    rpt.setLongMetric(A_ACK_UNEXPECTED, unstarted);
    return rpt;
  }

  public static SinkDecoBuilder builder() {
    return new SinkDecoBuilder() {
      @Override
      public EventSinkDecorator<EventSink> build(Context context,
          String... argv) {
        Preconditions.checkArgument(argv.length == 0, "usage: ackChecker");
        return new AckChecksumChecker<EventSink>(null, FlumeNode.getInstance()
            .getCollectorAckListener());
      }

    };
  }

} public class CustomDfsSink extends EventSink.Base {
  static final Logger LOG = LoggerFactory.getLogger(CustomDfsSink.class);

  private static final String A_OUTPUTFORMAT = "recordformat";

  boolean compressOutput;
  OutputFormat format;
  OutputStream writer;
  AtomicLong count = new AtomicLong();
  String path;
  Path dstPath;
  PathManager pathManager;

  public CustomDfsSink(String path, OutputFormat format) {
    Preconditions.checkArgument(path != null);
    Preconditions.checkArgument(format != null);
    this.path = path;
    this.format = format;
    this.writer = null;
  }

  @Override
  public void append(Event e) throws IOException, InterruptedException {
    if (writer == null) {
      throw new IOException("Append failed, did you open the writer?");
    }

    format.format(writer, e);
    count.getAndIncrement();
    super.append(e);
  }

  @Override
  public void close() throws IOException {
    LOG.info("Closing HDFS file: " + pathManager.getOpenPath());
    writer.flush();
    LOG.info("done writing raw file to hdfs");
    writer.close();
    pathManager.close();
    writer = null;
  }

  /**
   * Hadoop Compression Codecs that use Native libs require an instance of a
   * Configuration Object. They require this due to some check against knowing
   * weather or not the native libs have been loaded. GzipCodec, LzoCodec,
   * LzopCodec are all codecs that require Native libs. GZipCodec has a slight
   * exception that if native libs are not accessible it will use Pure Java.
   * This results in no errors just notices. BZip2Codec is an example codec that
   * doesn't use native libs.
   */
  @Override
  public void open() throws IOException {
    FlumeConfiguration conf = FlumeConfiguration.get();
    FileSystem hdfs;
    String codecName = conf.getCollectorDfsCompressCodec();
    CompressionCodec codec = getCodec(conf, codecName);

    if (codec == null) {
      dstPath = new Path(path);
      hdfs = dstPath.getFileSystem(conf);
      pathManager = new PathManager(hdfs, dstPath.getParent(), dstPath.getName());
      writer = pathManager.open();
      LOG.info("Creating HDFS file: " + pathManager.getOpenPath());
      return;
    }

    Compressor cmp = codec.createCompressor();
    dstPath = new Path(path + codec.getDefaultExtension());
    hdfs = dstPath.getFileSystem(conf);
    pathManager = new PathManager(hdfs, dstPath.getParent(), dstPath.getName());
    writer = pathManager.open();
    try {
      writer = codec.createOutputStream(writer, cmp);
    } catch (NullPointerException npe) {
      // tries to find "native" version of codec, if that fails, then tries to
      // find java version. If there is no java version, the createOutputStream
      // exits via NPE. We capture this and convert it into a IOE with a more
      // useful error message.
      LOG.error("Unable to load compression codec " + codec);
      throw new IOException("Unable to load compression codec " + codec);
    }
    LOG.info("Creating " + codec + " compressed HDFS file: "
        + pathManager.getOpenPath());
  }

  private static boolean codecMatches(Class<? extends CompressionCodec> cls,
      String codecName) {
    String simpleName = cls.getSimpleName();
    if (cls.getName().equals(codecName)
        || simpleName.equalsIgnoreCase(codecName)) {
      return true;
    }
    if (simpleName.endsWith("Codec")) {
      String prefix = simpleName.substring(0,
          simpleName.length() - "Codec".length());
      if (prefix.equalsIgnoreCase(codecName)) {
        return true;
      }
    }
    return false;
  }

  public static CompressionCodec getCodec(Configuration conf, String codecName) {
    List<Class<? extends CompressionCodec>> codecs = CompressionCodecFactory
        .getCodecClasses(FlumeConfiguration.get());
    // Wish we could base this on DefaultCodec but appears not all codec's
    // extend DefaultCodec(Lzo)
    CompressionCodec codec = null;
    ArrayList<String> codecStrs = new ArrayList<String>();
    codecStrs.add("None");
    for (Class<? extends CompressionCodec> cls : codecs) {
      codecStrs.add(cls.getSimpleName());

      if (codecMatches(cls, codecName)) {
        try {
          codec = cls.newInstance();
        } catch (InstantiationException e) {
          LOG.error("Unable to instantiate " + cls + " class");
        } catch (IllegalAccessException e) {
          LOG.error("Unable to access " + cls + " class");
        }
      }
    }

    if (codec == null) {
      if (!codecName.equalsIgnoreCase("None")) {
        throw new IllegalArgumentException("Unsupported compression codec "
            + codecName + ".  Please choose from: " + codecStrs);
      }
    } else if (codec instanceof Configurable) {
      // Must check instanceof codec as BZip2Codec doesn't inherit Configurable
      // Must set the configuration for Configurable objects that may or do use
      // native libs
      ((Configurable) codec).setConf(conf);
    }
    return codec;
  }

  public static SinkBuilder builder() {
    return new SinkBuilder() {
      @Override
      public EventSink create(Context context, Object... args) {
        if (args.length != 2 && args.length != 1) {
          // TODO (jon) make this message easier.
          throw new IllegalArgumentException(
              "usage: customdfs(\"[(hdfs|file|s3n|...)://namenode[:port]]/path\", format)");
        }

        Object format = (args.length == 1) ? null : args[1];
        OutputFormat fmt;
        try {
          fmt = FlumeBuilder.createFormat(FormatFactory.get(), format);
        } catch (FlumeSpecException e) {
          LOG.error("failed to load format " + format, e);
          throw new IllegalArgumentException("failed to load format " + format);
        }
        return new CustomDfsSink(args[0].toString(), fmt);
      }

      @Deprecated
      @Override
      public EventSink build(Context context, String... args) {
        // updated interface calls build(Context,Object...) instead
        throw new RuntimeException(
            "Old sink builder for CustomDfsSink should not be exercised");
      }
    };
  }

  @Override
  public String getName() {
    return "CustomDfs";
  }

  @Override
  public ReportEvent getMetrics() {
    ReportEvent rpt = super.getMetrics();
    rpt.setStringMetric(A_OUTPUTFORMAT, format.getBuilder().getName());
    rpt.setLongMetric(ReportEvent.A_COUNT, count.get());
    return rpt;
  }

  @Deprecated
  @Override
  public ReportEvent getReport() {
    ReportEvent rpt = super.getReport();
    rpt.setStringMetric(A_OUTPUTFORMAT, format.getBuilder().getName());
    rpt.setLongMetric(ReportEvent.A_COUNT, count.get());
    return rpt;
  }
} public class EscapedCustomDfsSink extends EventSink.Base {
  static final Logger LOG = LoggerFactory.getLogger(EscapedCustomDfsSink.class);
  final String path;
  OutputFormat format;

  CustomDfsSink writer = null;

  // We keep a - potentially unbounded - set of writers around to deal with
  // different tags on events. Therefore this feature should be used with some
  // care (where the set of possible paths is small) until we do something
  // more sensible with resource management.
  final Map<String, CustomDfsSink> sfWriters = new HashMap<String, CustomDfsSink>();

  // Used to short-circuit around doing regex matches when we know there are
  // no templates to be replaced.
  boolean shouldSub = false;
  private String filename = "";
  protected String absolutePath = "";

  public EscapedCustomDfsSink(String path, String filename, OutputFormat o) {
    this.path = path;
    this.filename = filename;
    shouldSub = Event.containsTag(path) || Event.containsTag(filename);
    this.format = o;
    absolutePath = path;
    if (filename != null && filename.length() > 0) {
      if (!absolutePath.endsWith(Path.SEPARATOR)) {
        absolutePath += Path.SEPARATOR;
      }
      absolutePath += this.filename;
    }
  }

  static protected OutputFormat getDefaultOutputFormat() {
    try {
      return FormatFactory.get().getOutputFormat(
          FlumeConfiguration.get().getDefaultOutputFormat());
    } catch (FlumeSpecException e) {
      LOG.warn("format from conf file not found, using default", e);
      return new RawOutputFormat();
    }
  }

  public EscapedCustomDfsSink(String path, String filename) {
    this(path, filename, getDefaultOutputFormat());
  }

  protected CustomDfsSink openWriter(String p) throws IOException {
    LOG.info("Opening " + p);
    CustomDfsSink w = new CustomDfsSink(p, format);
    w.open();
    return w;
  }

  /**
   * Writes the message to an HDFS file whose path is substituted with tags
   * drawn from the supplied event
   */
  @Override
  public void append(Event e) throws IOException, InterruptedException {
    CustomDfsSink w = writer;
    if (shouldSub) {
      String realPath = e.escapeString(absolutePath);
      w = sfWriters.get(realPath);
      if (w == null) {
        w = openWriter(realPath);
        sfWriters.put(realPath, w);
      }
    }
    w.append(e);
    super.append(e);
  }

  @Override
  public void close() throws IOException {
    if (shouldSub) {
      for (Entry<String, CustomDfsSink> e : sfWriters.entrySet()) {
        LOG.info("Closing " + e.getKey());
        e.getValue().close();
      }
    } else {
      LOG.info("Closing " + absolutePath);
      if (writer == null) {
        LOG.warn("EscapedCustomDfsSink's Writer for '" + absolutePath
            + "' was already closed!");
        return;
      }

      writer.close();
      writer = null;
    }
  }

  @Override
  public void open() throws IOException {
    if (!shouldSub) {
      writer = openWriter(absolutePath);
    }
  }

  public static SinkBuilder builder() {
    return new SinkBuilder() {
      @Override
      public EventSink create(Context context, Object... args) {
        Preconditions.checkArgument(args.length >= 1 && args.length <= 3,
            "usage: escapedCustomDfs(\"[(hdfs|file|s3n|...)://namenode[:port]]/path\""
                + "[, file [,outputformat ]])");

        String filename = "";
        if (args.length >= 2) {
          filename = args[1].toString();
        }

        Object format = FlumeConfiguration.get().getDefaultOutputFormat();
        if (args.length >= 3) {
          format = args[2];
        }

        OutputFormat o;
        try {
          o = FlumeBuilder.createFormat(FormatFactory.get(), format);
        } catch (FlumeSpecException e) {
          LOG.warn("Illegal format type " + format + ".", e);
          o = null;
        }
        Preconditions.checkArgument(o != null, "Illegal format type " + format
            + ".");

        return new EscapedCustomDfsSink(args[0].toString(), filename, o);
      }

      @Deprecated
      @Override
      public EventSink build(Context context, String... args) {
        // updated interface calls build(Context,Object...) instead
        throw new RuntimeException(
            "Old sink builder for EscapedCustomDfsSink should not be exercised");
      }

    };
  }
} public class ProcessTagger implements Tagger {

  private final static String DATE_FORMAT = "yyyyMMdd-HHmmssSSSZ";

  String lastTag;
  Date last;

  public ProcessTagger() {
  }

  @Override
  public String getTag() {
    return lastTag;
  }

  @Override
  public String newTag() {
    DateFormat dateFormat2 = new SimpleDateFormat(DATE_FORMAT);

    long pid = Thread.currentThread().getId();
    Date now = new Date(Clock.unixTime());
    long nanos = Clock.nanos();
    String f;
    // see: http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6231579
    synchronized (dateFormat2) {
      f = dateFormat2.format(now);
    }

    // formatted so that lexigraphical and chronological can use same sort
    // yyyyMMdd-HHmmssSSSz.0000000nanos.00000pid
    lastTag = String.format("%s.%012d.%08d", f, nanos, pid);

    this.last = now;
    return lastTag;
  }

  @Override
  public Date getDate() {
    if (last == null) {
      return new Date(0);
    }
    return new Date(last.getTime()); // Defensive copy
  }

} public class RollSink extends EventSink.Base {
  static final Logger LOG = LoggerFactory.getLogger(RollSink.class);
  public static final String C_TRIGGER = "trigger";

  final String fspec;
  EventSink curSink;
  final RollTrigger trigger;
  protected TriggerThread triggerThread = null;

  protected ExecutorService executor = null;

  private static int threadInitNumber = 0;
  final long checkLatencyMs; // default 4x a second
  private Context ctx; // roll context

  // reporting attributes and counters
  public final static String A_ROLLS = "rolls";
  public final static String A_ROLLFAILS = "rollfails";
  public final static String A_ROLLSPEC = "rollspec";
  public final String A_ROLL_TAG; // TODO (jon) parameterize this.
  public final static String DEFAULT_ROLL_TAG = "rolltag";

  final ReadWriteLock lock = new ReentrantReadWriteLock();

  final AtomicLong rolls = new AtomicLong();
  final AtomicLong rollfails = new AtomicLong();

  public RollSink(Context ctx, String spec, long maxAge, long checkMs) {
    this.ctx = ctx;
    A_ROLL_TAG = DEFAULT_ROLL_TAG;
    this.fspec = spec;
    this.trigger = new TimeTrigger(new ProcessTagger(), maxAge);
    this.checkLatencyMs = checkMs;
    LOG.info("Created RollSink: maxAge=" + maxAge + "ms trigger=[" + trigger
        + "] checkPeriodMs = " + checkLatencyMs + " spec='" + fspec + "'");
  }

  public RollSink(Context ctx, String spec, RollTrigger trigger, long checkMs) {
    this.ctx = ctx;
    A_ROLL_TAG = DEFAULT_ROLL_TAG;
    this.fspec = spec;
    this.trigger = trigger;
    this.checkLatencyMs = checkMs;
    LOG.info("Created RollSink: trigger=[" + trigger + "] checkPeriodMs = "
        + checkLatencyMs + " spec='" + fspec + "'");
  }

  private static synchronized int nextThreadNum() {
    return threadInitNumber++;
  }

  /**
   * This thread wakes up to check if a batch should be committed, no matter how
   * small it is.
   */
  class TriggerThread extends Thread {
    final CountDownLatch doneLatch = new CountDownLatch(1);
    final CountDownLatch startedLatch = new CountDownLatch(1);

    TriggerThread() {
      super("Roll-TriggerThread-" + nextThreadNum());
    }

    void doStart() {
      this.start();

      try {
        startedLatch.await();
      } catch (InterruptedException e) {
        LOG.warn("Interrupted while waiting for batch timeout thread to start");
      }
    }

    public void run() {
      startedLatch.countDown();
      try {
        while (!isInterrupted()) {
          // TODO there should probably be a lock on Roll sink but until we
          // handle
          // interruptions throughout the code, we cannot because this causes a
          // deadlock
          if (trigger.isTriggered()) {
            trigger.reset();

            LOG.debug("Rotate started by triggerthread... ");
            rotate();
            LOG.debug("Rotate stopped by triggerthread... ");
            continue;
          }

          try {
            Clock.sleep(checkLatencyMs);
          } catch (InterruptedException e) {
            LOG.debug("TriggerThread interrupted");
            doneLatch.countDown();
            return;
          }
        }
      } catch (InterruptedException e) {
        LOG.error("RollSink interrupted", e);
      }
      LOG.debug("TriggerThread shutdown");
      doneLatch.countDown();
    }
  };

  protected EventSink newSink(Context ctx) throws IOException {
    try {
      // TODO (jon) add roll-specific context information.
      return new CompositeSink(ctx, fspec);
    } catch (FlumeSpecException e) {
      // check done prior to construction.
      throw new IllegalArgumentException("This should never happen:"
          + e.getMessage());
    }
  };

  // currently it is assumed that there is only one thread handling appends, but
  // this has to be thread safe with respect to open and close (rotate) calls.
  Future<Void> future;

  // This is a large synchronized section. Won't fix until it becomes a problem.
  @Override
  public void append(final Event e) throws IOException, InterruptedException {
    Callable<Void> task = new Callable<Void>() {
      public Void call() throws Exception {
        synchronousAppend(e);
        return null;
      }
    };

    // keep track of the last future.
    future = executor.submit(task);
    try {
      future.get();
    } catch (ExecutionException e1) {
      Throwable cause = e1.getCause();
      if (cause instanceof IOException) {
        throw (IOException) cause;
      } else if (cause instanceof InterruptedException) {
        throw (InterruptedException) cause;
      } else if (cause instanceof RuntimeException) {
        throw (RuntimeException) cause;
      } else {
        // we have a throwable that is not an exception. (such as a
        // NoClassDefFoundError)
        LOG.error("Got a throwable that is not an exception! Bailing out!",
            e1.getCause());
        throw new RuntimeException(e1.getCause());
      }
    } catch (CancellationException ce) {
      Thread.currentThread().interrupt();
      throw new InterruptedException(
          "Blocked append interrupted by rotation event");
    } catch (InterruptedException ex) {
      LOG.warn("Unexpected Exception " + ex.getMessage(), ex);
      Thread.currentThread().interrupt();
      throw (InterruptedException) ex;
    }
  }

  public void synchronousAppend(Event e) throws IOException,
      InterruptedException {
    Preconditions.checkState(curSink != null,
        "Attempted to append when rollsink not open");

    if (trigger.isTriggered()) {
      trigger.reset();
      LOG.debug("Rotate started by append... ");
      rotate();
      LOG.debug("... rotate completed by append.");
    }
    String tag = trigger.getTagger().getTag();

    e.set(A_ROLL_TAG, tag.getBytes());
    lock.readLock().lock();
    try {
      curSink.append(e);
      trigger.append(e);
      super.append(e);
    } finally {
      lock.readLock().unlock();
    }
  }

  /**
   * This method assumes it will be guarded by locks
   */
  boolean synchronousRotate() throws InterruptedException, IOException {
    rolls.incrementAndGet();
    if (curSink == null) {
      // wtf, was closed or never opened
      LOG.error("Attempting to rotate an already closed roller");
      return false;
    }
    try {
      curSink.close();
    } catch (IOException ioe) {
      // Eat this exception and just move to reopening
      LOG.warn("IOException when closing subsink", ioe);

      // other exceptions propagate out of here.
    }
    curSink = newSink(ctx);
    curSink.open();
    LOG.debug("rotated sink ");
    return true;
  }

  public boolean rotate() throws InterruptedException {
    while (!lock.writeLock().tryLock(1000, TimeUnit.MILLISECONDS)) {
      // interrupt the future on the other.
      if (future != null) {
        future.cancel(true);
      }

      // NOTE: there is no guarantee that this cancel actually succeeds.
    }
    // interrupted, lets go.
    try {
      synchronousRotate();
    } catch (IOException e1) {
      // TODO This is an error condition that needs to be handled -- could be
      // due to resource exhaustion.
      LOG.error("Failure when attempting to rotate and open new sink: "
          + e1.getMessage());
      rollfails.incrementAndGet();
      return false;
    } finally {
      lock.writeLock().unlock();
    }
    return true;
  }

  @Override
  public void close() throws IOException, InterruptedException {
    LOG.info("closing RollSink '" + fspec + "'");

    // attempt to get the lock, and if we cannot, issue a cancel
    while (!lock.writeLock().tryLock(1000, TimeUnit.MILLISECONDS)) {
      // interrupt the future on the other.
      if (future != null) {
        future.cancel(true);
      }
    }

    // we have the write lock now.
    try {
      if (executor != null) {
        executor.shutdown();
        executor = null;
      }
    } finally {
      lock.writeLock().unlock();
    }
    // TODO triggerThread can race with an open call, but we really need to
    // avoid having the await while locked!
    if (triggerThread != null) {
      try {
        triggerThread.interrupt();
        triggerThread.doneLatch.await();
      } catch (InterruptedException e) {
        LOG.warn("Interrupted while waiting for batch timeout thread to finish");
        // TODO check finally
        throw e;
      }
    }

    lock.writeLock().lock();
    try {
      if (curSink == null) {
        LOG.info("double close '" + fspec + "'");
        return;
      }
      curSink.close();
      curSink = null;
    } finally {
      lock.writeLock().unlock();
    }
  }

  @Override
  public void open() throws IOException, InterruptedException {
    lock.writeLock().lock();
    try {
      if (executor != null) {
        throw new IllegalStateException(
            "Attempting to open already open roll sink");
      }
      executor = Executors.newFixedThreadPool(1);

      Preconditions.checkState(curSink == null,
          "Attempting to open already open RollSink '" + fspec + "'");
      LOG.info("opening RollSink  '" + fspec + "'");
      trigger.getTagger().newTag();
      triggerThread = new TriggerThread();
      triggerThread.doStart();

      try {
        curSink = newSink(ctx);
        curSink.open();
      } catch (IOException e1) {
        LOG.warn("Failure when attempting to open initial sink", e1);
        executor.shutdown();
        executor = null;
      }
    } finally {
      lock.writeLock().unlock();
    }
  }

  @Override
  public String getName() {
    return "Roll";
  }

  @Override
  synchronized public ReportEvent getMetrics() {
    ReportEvent rpt = super.getMetrics();
    rpt.setLongMetric(A_ROLLS, rolls.get());
    rpt.setLongMetric(A_ROLLFAILS, rollfails.get());
    rpt.setStringMetric(A_ROLLSPEC, fspec);
    return rpt;
  }

  @Override
  public Map<String, Reportable> getSubMetrics() {
    // subReports will handle case where curSink is null
    return ReportUtil.subReports(curSink);
  }

  public String getRollSpec() {
    return fspec;
  }

  @Deprecated
  @Override
  synchronized public ReportEvent getReport() {
    ReportEvent rpt = super.getReport();
    rpt.setLongMetric(A_ROLLS, rolls.get());
    rpt.setLongMetric(A_ROLLFAILS, rollfails.get());
    rpt.setStringMetric(A_ROLLSPEC, fspec);
    return rpt;
  }

  @Deprecated
  @Override
  public void getReports(String namePrefix, Map<String, ReportEvent> reports) {
    super.getReports(namePrefix, reports);
    if (curSink != null) {
      curSink.getReports(namePrefix + getName() + ".", reports);
    }
  }

  /**
   * This is currently only used in tests.
   */
  @Deprecated
  public String getCurrentTag() {
    return trigger.getTagger().getTag();
  }

  public static RollTrigger createRollTrigger(Context ctx, long rollmillis) {
    RollTrigger rt = null;
    FunctionSpec fs = ctx.getObj(C_TRIGGER, FunctionSpec.class);
    if (fs == null) {
      rt = new TimeTrigger(new ProcessTagger(), rollmillis);
    } else {
      if ("time".equals(fs.getName())) {
        rt = new TimeTrigger(new ProcessTagger(), rollmillis);
      } else if ("size".equals(fs.getName())) {
        Tagger t = new ProcessTagger();
        Preconditions.checkArgument(fs.getArgs().length == 1,
            "size trigger requires argument");
        long sz = Long.parseLong(fs.getArgs()[0].toString());
        rt = new OrTrigger(t, new TimeTrigger(t, rollmillis), new SizeTrigger(
            sz, t));
      } else {
        throw new IllegalArgumentException("Illegal trigger type specified: "
            + fs.getName());
      }
    }
    return rt;
  }

  /**
   * Builder for a spec based rolling sink. (most general version, does not
   * necessarily output to files!).
   *
   * Note: trigger is in context so can be set from any parent source such as
   * collectorSink
   */
  public static SinkBuilder builder() {
    return new SinkBuilder() {
      @Override
      public EventSink create(Context ctx, Object... argv) {
        Preconditions.checkArgument(argv.length >= 2 && argv.length <= 3,
            "roll(rollmillis[, checkmillis]{, trigger=time|size(n)}) { sink }");
        String spec = argv[0].toString();
        long rollmillis = Long.parseLong(argv[1].toString());

        long checkmillis = 250;
        if (argv.length >= 3) {
          checkmillis = Long.parseLong(argv[2].toString());
        }

        RollTrigger rt = createRollTrigger(ctx, rollmillis);

        try {
          // check sub spec to make sure it works.
          FlumeBuilder.buildSink(ctx, spec);

          // ok it worked, instantiate the roller
          return new RollSink(ctx, spec, rt, checkmillis);
        } catch (FlumeSpecException e) {
          throw new IllegalArgumentException("Failed to parse/build " + spec, e);
        }
      }

      @Deprecated
      @Override
      public EventSink build(Context ctx, String... argv) {
        throw new RuntimeException(
            "Old sink builder for RollSink should not be exercised");
      }
    };
  }
} public interface Tagger {

  public final static String A_TXID = "txid"; // "transaction id"

  String getTag();

  /**
   * Our sinks often rotate and each version needs a separate unique tag/name.
   * This method updates state and creates the next name.
   */
  String newTag();

  Date getDate();

}
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.log4j.Level;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.agent.FlumeNode;
import com.cloudera.flume.agent.durability.NaiveFileWALDeco;
import com.cloudera.flume.agent.durability.WALManager;
import com.cloudera.flume.conf.Context;
import com.cloudera.flume.conf.FlumeArgException;
import com.cloudera.flume.conf.FlumeBuilder;
import com.cloudera.flume.conf.FlumeConfiguration;
import com.cloudera.flume.conf.FlumeSpecException;
import com.cloudera.flume.conf.LogicalNodeContext;
import com.cloudera.flume.conf.ReportTestingContext;
import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
import com.cloudera.flume.conf.SinkFactoryImpl;
import com.cloudera.flume.core.Event;
import com.cloudera.flume.core.EventImpl;
import com.cloudera.flume.core.EventSink;
import com.cloudera.flume.core.EventSource;
import com.cloudera.flume.core.EventUtil;
import com.cloudera.flume.handlers.debug.LazyOpenDecorator;
import com.cloudera.flume.handlers.debug.MemorySinkSource;
import com.cloudera.flume.handlers.debug.NoNlASCIISynthSource;
import com.cloudera.flume.handlers.endtoend.AckChecksumChecker;
import com.cloudera.flume.handlers.endtoend.AckChecksumInjector;
import com.cloudera.flume.handlers.endtoend.AckListener;
import com.cloudera.flume.handlers.hdfs.CustomDfsSink;
import com.cloudera.flume.handlers.hdfs.EscapedCustomDfsSink;
import com.cloudera.flume.handlers.rolling.ProcessTagger;
import com.cloudera.flume.handlers.rolling.RollSink;
import com.cloudera.flume.handlers.rolling.Tagger;
import com.cloudera.flume.handlers.thrift.Priority;
import com.cloudera.flume.handlers.thrift.ThriftFlumeEvent;
import com.cloudera.flume.reporter.ReportEvent;
import com.cloudera.flume.reporter.ReportManager;
import com.cloudera.util.FlumeTestHarness;
import com.cloudera.util.Clock;
import com.cloudera.util.FileUtil;
import com.cloudera.util.Pair;
/**
 * Licensed to Cloudera, Inc. under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  Cloudera, Inc. licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.cloudera.flume.collector;





/**
 * This tests the builder and makes sure we can close a collector properly when
 * interrupted.
 * 
 * TODO This should, but does not, test situations where the collectorSink
 * actually connects to an HDFS namenode, and then recovers from when an actual
 * HDFS goes down and comes back up. Instead this contains tests that shows when
 * a HDFS connection is fails, the retry metchanisms are forced to exit.
 */
public class TestCollectorSink {
  final static Logger LOG = LoggerFactory.getLogger(TestCollectorSink.class);

  @Before
  public void setUp() {
    // Log4j specific settings for debugging.
    org.apache.log4j.Logger.getLogger(CollectorSink.class)
        .setLevel(Level.DEBUG);
    org.apache.log4j.Logger.getLogger(AckChecksumChecker.class).setLevel(
        Level.DEBUG);
    org.apache.log4j.Logger.getLogger(CustomDfsSink.class).setLevel(Level.WARN);
    org.apache.log4j.Logger.getLogger(EscapedCustomDfsSink.class).setLevel(
        Level.WARN);
  }

  @Test
  public void testBuilder() throws FlumeSpecException {
    Exception ex = null;
    try {
      String src = "collectorSink";
      FlumeBuilder.buildSink(new Context(), src);
    } catch (Exception e) {
      return;
    }
    assertNotNull("No exception thrown!", ex != null);

    // dir / filename
    String src2 = "collectorSink(\"file:///tmp/test\", \"testfilename\")";
    FlumeBuilder.buildSink(new Context(), src2);

    // millis
    String src3 = "collectorSink(\"file:///tmp/test\", \"testfilename\", 1000)";
    FlumeBuilder.buildSink(new Context(), src3);

    // format
    String src4 = "collectorSink(\"file:///tmp/test\", \"testfilename\", 1000, avro)";
    FlumeBuilder.buildSink(new Context(), src4);

    // format
    src4 = "collectorSink(\"file:///tmp/test\", \"testfilename\", 1000, seqfile)";
    FlumeBuilder.buildSink(new Context(), src4);

    // format
    src4 = "collectorSink(\"file:///tmp/test\", \"testfilename\", 1000, seqfile(\"bzip2\"))";
    FlumeBuilder.buildSink(new Context(), src4);

  }

  @Test(expected = FlumeArgException.class)
  public void testBuilderFail() throws FlumeSpecException {
    // too many arguments
    String src4 = "collectorSink(\"file:///tmp/test\", \"bkjlasdf\", 1000, avro, 1231)";
    FlumeBuilder.buildSink(new Context(), src4);
  }

  /**
   * play evil games with escape characters, and check that that old syntax
   * works with old semantics.
   */
  @Test
  public void testBuilderHandleEvilSlashes() throws FlumeSpecException {

    String src4 = "collectorSink(\"file://C:\\tmp\\test\", \"file\", 1000)";
    CollectorSink snk = (CollectorSink) FlumeBuilder.buildSink(new Context(),
        src4);
    assertEquals(
        "escapedCustomDfs(\"file://C:\\tmp\\test\",\"file%{rolltag}\" )",
        snk.roller.getRollSpec());
  }

  @Test
  public void testOpenClose() throws FlumeSpecException, IOException,
      InterruptedException {
    String src2 = "collectorSink(\"file:///tmp/test\",\"testfilename\")";

    for (int i = 0; i < 100; i++) {
      EventSink snk = FlumeBuilder.buildSink(new Context(), src2);
      snk.open();
      snk.close();
    }
  }

  /**
   * Test that file paths are correctly constructed from dir + path + tag
   * 
   * @throws InterruptedException
   * @throws FlumeSpecException
   */
  @Test
  public void testCorrectFilename() throws IOException, InterruptedException,
      FlumeSpecException {
    CollectorSink sink = new CollectorSink(new Context(),
        "file:///tmp/flume-test-correct-filename", "actual-file-", 10000,
        new Tagger() {
          public String getTag() {
            return "tag";
          }

          public String newTag() {
            return "tag";
          }

          public Date getDate() {
            return new Date();
          }

          public void annotate(Event e) {

          }
        }, 250, FlumeNode.getInstance().getCollectorAckListener());