From 075d8cbc4c1b7247e977451b31590f3beea45157 Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Mon, 10 Jan 2011 15:45:46 -0800
Subject: [PATCH] FLUME-165: Allow collector sink to output to other sinks

---
 src/antlr/com/cloudera/flume/conf/FlumeDeploy.g    |    5 +
 .../cloudera/flume/collector/CollectorSink.java    |   87 ++++++++++++++-----
 src/java/com/cloudera/flume/conf/FlumeBuilder.java |   39 ++++++++-
 src/java/com/cloudera/flume/conf/FlumeSpecGen.java |   18 ++++
 .../com/cloudera/flume/conf/SinkFactoryImpl.java   |    9 +-
 .../cloudera/flume/handlers/rolling/RollSink.java  |    4 +
 .../flume/collector/TestCollectorSink.java         |   71 ++++++++++++----
 .../com/cloudera/flume/conf/TestFlumeBuilder.java  |   26 ++++++
 .../com/cloudera/flume/conf/TestFlumeSpecGen.java  |   12 +++
 .../com/cloudera/flume/conf/TestParser.java        |   23 +++++-
 10 files changed, 245 insertions(+), 49 deletions(-)

diff --git a/src/antlr/com/cloudera/flume/conf/FlumeDeploy.g b/src/antlr/com/cloudera/flume/conf/FlumeDeploy.g
index 9868bb6..429acb5 100644
--- a/src/antlr/com/cloudera/flume/conf/FlumeDeploy.g
+++ b/src/antlr/com/cloudera/flume/conf/FlumeDeploy.g
@@ -28,6 +28,7 @@ tokens {
   SINK;
   BACKUP;
   ROLL;
+  GEN;
   DECO;
   SOURCE;
   MULTI;
@@ -120,6 +121,7 @@ simpleSink	:	'[' multiSink ']'  	-> ^(MULTI multiSink)
 		|	'{' decoratedSink '}'	-> ^(DECO decoratedSink)
 		|	'<' failoverSink '>'	-> ^(BACKUP failoverSink)
         |   rollSink                -> rollSink
+        |   genCollectorSink        -> genCollectorSink
 		; 
  			
 
@@ -128,6 +130,9 @@ multiSink       :  simpleSink (',' simpleSink)* 	-> simpleSink* ;
 failoverSink    :  simpleSink ('?' simpleSink)+	-> simpleSink+;	
 rollSink        :  'roll' args '{' simpleSink '}'
                                   -> ^(ROLL simpleSink args);
+genCollectorSink       :  'collector' args '{' simpleSink '}'
+                                  -> ^(GEN 'collector' simpleSink args?);
+
 
 args    : '(' ( arglist (',' kwarglist)?  ) ')' -> arglist kwarglist?
         | '(' kwarglist ')' -> kwarglist? 
diff --git a/src/java/com/cloudera/flume/collector/CollectorSink.java b/src/java/com/cloudera/flume/collector/CollectorSink.java
index 99f89cd..ed5bc46 100644
--- a/src/java/com/cloudera/flume/collector/CollectorSink.java
+++ b/src/java/com/cloudera/flume/collector/CollectorSink.java
@@ -25,7 +25,7 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
-import org.apache.hadoop.fs.Path;
+import org.apache.commons.lang.StringEscapeUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,6 +34,7 @@ import com.cloudera.flume.conf.Context;
 import com.cloudera.flume.conf.FlumeConfiguration;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
+import com.cloudera.flume.core.CompositeSink;
 import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.core.EventSinkDecorator;
@@ -43,7 +44,6 @@ import com.cloudera.flume.handlers.debug.InsistentOpenDecorator;
 import com.cloudera.flume.handlers.debug.StubbornAppendSink;
 import com.cloudera.flume.handlers.endtoend.AckChecksumChecker;
 import com.cloudera.flume.handlers.endtoend.AckListener;
-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;
@@ -55,13 +55,11 @@ import com.cloudera.util.CumulativeCappedExponentialBackoff;
 import com.google.common.base.Preconditions;
 
 /**
- * This collector sink is the high level specification a user would use.
- * 
- * Currently is is just a DFSEventSink that uses some default values from the
- * flume configuration file and is overridable by the user.
- * 
- * TODO (jon) this default output needs to be replaced with a combo of henry's
- * new tagged output writer and the custom format dfs writer..
+ * This collector sink is the high level specification a user would use. The
+ * subsink spec allows for specifying batching, gunzip, multiple sinks or
+ * whatever else a user would want to specify. The key feature with any sub sink
+ * specified of this is that it is part of a roller, and that any accrued acks
+ * are send when this roller closes the subsink.
  */
 public class CollectorSink extends EventSink.Base {
   static final Logger LOG = LoggerFactory.getLogger(CollectorSink.class);
@@ -69,6 +67,7 @@ public class CollectorSink extends EventSink.Base {
   final EventSink snk;
   AckAccumulator accum = new AckAccumulator();
   final AckListener ackDest;
+  final String snkSpec;
 
   // This is a container for acks that should be ready for delivery when the
   // hdfs sink is closed/flushed
@@ -77,23 +76,29 @@ public class CollectorSink extends EventSink.Base {
   // References package exposed for testing
   final RollSink roller;
 
-  CollectorSink(Context ctx, String path, String filename, long millis,
-      AckListener ackDest) throws FlumeSpecException {
-    this(ctx, path, filename, millis, new ProcessTagger(), 250, ackDest);
+  CollectorSink(Context ctx, String snkSpec, long millis, AckListener ackDest)
+      throws FlumeSpecException {
+    this(ctx, snkSpec, millis, new ProcessTagger(), 250, ackDest);
   }
 
-  CollectorSink(Context ctx, final String logdir, final String filename,
-      final long millis, final Tagger tagger, long checkmillis,
-      AckListener ackDest) {
+  CollectorSink(Context ctx, final String snkSpec, final long millis,
+      final Tagger tagger, long checkmillis, AckListener ackDest) {
     this.ackDest = ackDest;
-    this.roller = new RollSink(ctx, "collectorSink", new TimeTrigger(tagger,
-        millis), checkmillis) {
+    this.snkSpec = snkSpec;
+    roller = new RollSink(ctx, snkSpec, new TimeTrigger(tagger, millis),
+        checkmillis) {
+      // this is wraps the normal roll sink with an extra roll detection
+      // decorator that triggers ack delivery on close.
       @Override
       public EventSink newSink(Context ctx) throws IOException {
         String tag = tagger.newTag();
-        String path = logdir + Path.SEPARATOR_CHAR;
-        EventSink dfs = new EscapedCustomDfsSink(path, filename + tag);
-        return new RollDetectDeco(dfs, tag);
+        EventSink drain;
+        try {
+          drain = new CompositeSink(ctx, snkSpec);
+        } catch (FlumeSpecException e) {
+          throw new IOException("Unable to instantiate '" + snkSpec + "'", e);
+        }
+        return new RollDetectDeco(drain, tag);
       }
     };
 
@@ -112,7 +117,7 @@ public class CollectorSink extends EventSink.Base {
 
     // { ackChecksumChecker => insistentAppend => stubbornAppend =>
     // insistentOpen => mask("rolltag") => roll(xx) { rollDetect =>
-    // escapedCusomtDfs } }
+    // subsink } }
     EventSink tmp = new MaskDecorator<EventSink>(roller, "rolltag");
     tmp = new InsistentOpenDecorator<EventSink>(tmp, backoff1);
     tmp = new StubbornAppendSink<EventSink>(tmp);
@@ -120,6 +125,17 @@ public class CollectorSink extends EventSink.Base {
     snk = new AckChecksumChecker<EventSink>(tmp, accum);
   }
 
+  /**
+   * This is a compatibility mode for older version of the tests
+   */
+  CollectorSink(Context ctx, String path, String filename, long millis,
+      final Tagger tagger, long checkmillis, AckListener ackDest)
+      throws FlumeSpecException {
+    this(ctx, "escapedCustomDfs(\"" + StringEscapeUtils.escapeJava(path)
+        + "\",\"" + StringEscapeUtils.escapeJava(filename) + "%{rolltag}"
+        + "\" )", millis, tagger, checkmillis, ackDest);
+  }
+
   String curRollTag;
 
   /**
@@ -165,7 +181,6 @@ public class CollectorSink extends EventSink.Base {
         master.end(at);
       }
     }
-
   };
 
   /**
@@ -246,6 +261,31 @@ public class CollectorSink extends EventSink.Base {
     return new SinkBuilder() {
       @Override
       public EventSink build(Context context, String... argv) {
+        Preconditions.checkArgument(argv.length <= 2 && argv.length >= 0,
+            "usage: collector[(rollmillis)] { subsink }");
+        String snkSpec = argv[0];
+
+        long millis = FlumeConfiguration.get().getCollectorRollMillis();
+        if (argv.length >= 2) {
+          millis = Long.parseLong(argv[1]);
+        }
+        try {
+          EventSink deco = new CollectorSink(context, snkSpec, millis,
+              FlumeNode.getInstance().getCollectorAckListener());
+          return deco;
+        } catch (FlumeSpecException e) {
+          LOG.error("CollectorDecorator spec error " + e, e);
+          throw new IllegalArgumentException(
+              "usage: collector[(rollmillis)] { subsink }" + e);
+        }
+      }
+    };
+  }
+
+  public static SinkBuilder hdfsBuilder() {
+    return new SinkBuilder() {
+      @Override
+      public EventSink build(Context context, String... argv) {
         Preconditions.checkArgument(argv.length <= 3 && argv.length >= 2,
             "usage: collectorSink[(dfsdir,path[,rollmillis])]");
         String logdir = FlumeConfiguration.get().getCollectorDfsDir(); // default
@@ -262,7 +302,8 @@ public class CollectorSink extends EventSink.Base {
         }
         try {
           EventSink snk = new CollectorSink(context, logdir, prefix, millis,
-              FlumeNode.getInstance().getCollectorAckListener());
+              new ProcessTagger(), 250, FlumeNode.getInstance()
+                  .getCollectorAckListener());
           return snk;
         } catch (FlumeSpecException e) {
           LOG.error("CollectorSink spec error " + e, e);
diff --git a/src/java/com/cloudera/flume/conf/FlumeBuilder.java b/src/java/com/cloudera/flume/conf/FlumeBuilder.java
index 98e1ba8..24baaff 100644
--- a/src/java/com/cloudera/flume/conf/FlumeBuilder.java
+++ b/src/java/com/cloudera/flume/conf/FlumeBuilder.java
@@ -34,6 +34,8 @@ import org.apache.commons.lang.StringEscapeUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.cloudera.flume.collector.CollectorSink;
+import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
 import com.cloudera.flume.core.BackOffFailOverSink;
 import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventSink;
@@ -89,7 +91,7 @@ public class FlumeBuilder {
     DEC, HEX, OCT, STRING, BOOL, FLOAT, // literals
     SINK, SOURCE, // sink or source
     KWARG, // kwarg support
-    MULTI, DECO, BACKUP, LET, ROLL, FAILCHAIN, // compound sinks
+    MULTI, DECO, BACKUP, LET, ROLL, GEN, FAILCHAIN, // compound sinks
     NODE, // combination of sink and source
   };
 
@@ -430,7 +432,7 @@ public class FlumeBuilder {
       SinkFactory sinkFactory) throws FlumeSpecException {
     ASTNODE type = ASTNODE.valueOf(t.getText()); // convert to enum
     switch (type) {
-    case SINK:
+    case SINK: {
 
       Context ctx = new Context(context);
       Pair<String, List<String>> idargs = handleArgs(t, ctx);
@@ -445,7 +447,7 @@ public class FlumeBuilder {
             + FlumeSpecGen.genEventSink(t));
       }
       return snk;
-
+    }
     case MULTI: {
       List<CommonTree> elems = (List<CommonTree>) t.getChildren();
       List<EventSink> snks = new ArrayList<EventSink>();
@@ -584,6 +586,37 @@ public class FlumeBuilder {
 
     }
 
+    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<String>> idArgs = handleArgs(t, ctx);
+        String sourceType = idArgs.getLeft();
+        List<String> args = idArgs.getRight();
+        args.add(0, body);
+
+        // TODO replace with Generator Sink lookup
+        Preconditions.checkArgument("collector".equals(sourceType));
+        SinkBuilder builder = CollectorSink.builder();
+
+        return builder.build(ctx, args.toArray(new String[0]));
+
+      } 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
diff --git a/src/java/com/cloudera/flume/conf/FlumeSpecGen.java b/src/java/com/cloudera/flume/conf/FlumeSpecGen.java
index 3462d91..a2ee2da 100644
--- a/src/java/com/cloudera/flume/conf/FlumeSpecGen.java
+++ b/src/java/com/cloudera/flume/conf/FlumeSpecGen.java
@@ -197,6 +197,24 @@ public class FlumeSpecGen {
       return "failchain" + argSink + " { " + bodySink + " }";
 
     }
+
+    case GEN: {
+      List<CommonTree> genNodes = (List<CommonTree>) t.getChildren();
+      Preconditions.checkArgument(genNodes.size() >= 2);
+      String genType = genNodes.get(0).getText();
+      CommonTree body = genNodes.get(1);
+      String bodySink = genEventSink(body);
+
+      List<String> rargs = new ArrayList<String>();
+      for (int i = 2; i < genNodes.size(); i++) {
+        CommonTree arg = genNodes.get(i);
+        rargs.add(genArg(arg));
+      }
+      String argSink = genArgs(rargs, "(", ",", ")");
+
+      return genType + argSink + " { " + bodySink + " }";
+
+    }
       // TODO (jon) handle pattern match splitter
 
     default:
diff --git a/src/java/com/cloudera/flume/conf/SinkFactoryImpl.java b/src/java/com/cloudera/flume/conf/SinkFactoryImpl.java
index 7a9c6be..426aa2f 100644
--- a/src/java/com/cloudera/flume/conf/SinkFactoryImpl.java
+++ b/src/java/com/cloudera/flume/conf/SinkFactoryImpl.java
@@ -97,7 +97,7 @@ public class SinkFactoryImpl extends SinkFactory {
   // The actual types are <String, SinkBuilder>
   static Object[][] sinkList = {
       // high level sinks.
-      { "collectorSink", CollectorSink.builder() },
+      { "collectorSink", CollectorSink.hdfsBuilder() },
 
       { "agentSink", AgentSink.e2eBuilder() },
       { "agentE2ESink", AgentSink.e2eBuilder() }, // now with acks
@@ -129,7 +129,8 @@ public class SinkFactoryImpl extends SinkFactory {
       { "dfs", DFSEventSink.builder() }, // escapes
       { "customdfs", CustomDfsSink.builder() }, // does not escape
       { "escapedCustomDfs", EscapedCustomDfsSink.builder() }, // escapes
-      { "rpcSink", RpcSink.builder() }, //creates AvroEventSink or ThriftEventSink
+      { "rpcSink", RpcSink.builder() }, // creates AvroEventSink or
+      // ThriftEventSink
       { "syslogTcp", SyslogTcpSink.builder() },
       { "irc", IrcSink.builder() },
       { "thriftSink", ThriftEventSink.builder() },
@@ -202,9 +203,9 @@ public class SinkFactoryImpl extends SinkFactory {
       { "bloomGen", BloomGeneratorDeco.builder() },
       { "bloomCheck", BloomCheckDecorator.builder() },
       { "mult", MultiplierDecorator.builder() },
-      { "delay", DelayDecorator.builder() }, 
+      { "delay", DelayDecorator.builder() },
       { "choke", ChokeDecorator.builder() },
-  
+
   };
 
   Map<String, SinkBuilder> sinks = new HashMap<String, SinkBuilder>();
diff --git a/src/java/com/cloudera/flume/handlers/rolling/RollSink.java b/src/java/com/cloudera/flume/handlers/rolling/RollSink.java
index 5dffa21..f5e6e4f 100644
--- a/src/java/com/cloudera/flume/handlers/rolling/RollSink.java
+++ b/src/java/com/cloudera/flume/handlers/rolling/RollSink.java
@@ -266,6 +266,10 @@ public class RollSink extends EventSink.Base {
     return ReportUtil.subReports(curSink);
   }
 
+  public String getRollSpec() {
+    return fspec;
+  }
+
   @Deprecated
   @Override
   synchronized public ReportEvent getReport() {
diff --git a/src/javatest/com/cloudera/flume/collector/TestCollectorSink.java b/src/javatest/com/cloudera/flume/collector/TestCollectorSink.java
index 4b5567a..f00c069 100644
--- a/src/javatest/com/cloudera/flume/collector/TestCollectorSink.java
+++ b/src/javatest/com/cloudera/flume/collector/TestCollectorSink.java
@@ -20,7 +20,6 @@ package com.cloudera.flume.collector;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
 
 import java.io.File;
@@ -41,13 +40,14 @@ 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.FlumeSpecException;
 import com.cloudera.flume.conf.LogicalNodeContext;
+import com.cloudera.flume.conf.ReportTestingContext;
 import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventImpl;
 import com.cloudera.flume.core.EventSink;
-import com.cloudera.flume.core.EventSinkDecorator;
 import com.cloudera.flume.core.EventSource;
 import com.cloudera.flume.core.EventUtil;
 import com.cloudera.flume.handlers.debug.LazyOpenDecorator;
@@ -58,8 +58,11 @@ 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.reporter.ReportEvent;
+import com.cloudera.flume.reporter.ReportManager;
 import com.cloudera.util.BenchmarkHarness;
 import com.cloudera.util.Clock;
 import com.cloudera.util.FileUtil;
@@ -80,8 +83,8 @@ public class TestCollectorSink {
   @Before
   public void setUp() {
     // Log4j specific settings for debugging.
-    org.apache.log4j.Logger.getLogger(CollectorSink.class)
-        .setLevel(Level.DEBUG);
+    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);
@@ -107,15 +110,28 @@ public class TestCollectorSink {
     // millis
     String src3 = "collectorSink(\"file:///tmp/test\", \"testfilename\", 1000)";
     FlumeBuilder.buildSink(new Context(), src3);
+  }
 
-    try {
-      // too many arguments
-      String src4 = "collectorSink(\"file:///tmp/test\", \"bkjlasdf\", 1000, 1000)";
-      FlumeBuilder.buildSink(new Context(), src4);
-    } catch (Exception e) {
-      return;
-    }
-    fail("unexpected fall through");
+  @Test(expected = FlumeArgException.class)
+  public void testBuilderFail() throws FlumeSpecException {
+    // too many arguments
+    String src4 = "collectorSink(\"file:///tmp/test\", \"bkjlasdf\", 1000, 1000)";
+    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
@@ -134,9 +150,11 @@ public class TestCollectorSink {
    * Test that file paths are correctly constructed from dir + path + tag
    * 
    * @throws InterruptedException
+   * @throws FlumeSpecException
    */
   @Test
-  public void testCorrectFilename() throws IOException, InterruptedException {
+  public void testCorrectFilename() throws IOException, InterruptedException,
+      FlumeSpecException {
     CollectorSink sink = new CollectorSink(new Context(),
         "file:///tmp/flume-test-correct-filename", "actual-file-", 10000,
         new Tagger() {
@@ -209,8 +227,8 @@ public class TestCollectorSink {
     // thus closes hdfs handles).
     String snkspec = "collectorSink(\"file:///" + tmpdir.getAbsolutePath()
         + "\",\"\")";
-    CollectorSink coll = (CollectorSink) FlumeBuilder.buildSink(new Context(),
-        snkspec);
+    CollectorSink coll = (CollectorSink) FlumeBuilder.buildSink(
+        new Context(), snkspec);
     AckChecksumChecker<EventSink> chk = (AckChecksumChecker<EventSink>) coll
         .getSink();
     // insistent append
@@ -600,13 +618,15 @@ public class TestCollectorSink {
           startCount++;
         }
 
-      };
+      }
+      ;
 
       TestAckListener fakeMasterRpc = new TestAckListener();
 
       // massive roll millis because the test will force fast and frequent rolls
-      CollectorSink cs = new CollectorSink(LogicalNodeContext.testingContext(),
-          "file:///" + dir.getAbsolutePath(), "test", 1000000, fakeMasterRpc) {
+      CollectorSink cs = new CollectorSink(LogicalNodeContext
+          .testingContext(), "file:///" + dir.getAbsolutePath(), "test",
+          1000000, new ProcessTagger(), 250, fakeMasterRpc) {
         @Override
         public void append(Event e) throws IOException, InterruptedException {
           LOG.info("Pre  append: "
@@ -657,4 +677,19 @@ public class TestCollectorSink {
       FileUtil.rmr(dir);
     }
   }
+
+  @Test
+  public void testMultipleSinks() throws FlumeSpecException, IOException,
+      InterruptedException {
+    String spec = "collector(5000) { [ counter(\"foo\"), counter(\"bar\") ] }";
+    EventSink snk = FlumeBuilder.buildSink(new ReportTestingContext(
+        LogicalNodeContext.testingContext()), spec);
+    snk.open();
+    snk.append(new EventImpl("this is a test".getBytes()));
+    snk.close();
+    ReportEvent rpta = ReportManager.get().getReportable("foo").getMetrics();
+    assertEquals(1, (long) rpta.getLongMetric("foo"));
+    ReportEvent rptb = ReportManager.get().getReportable("bar").getMetrics();
+    assertEquals(1, (long) rptb.getLongMetric("bar"));
+  }
 }
diff --git a/src/javatest/com/cloudera/flume/conf/TestFlumeBuilder.java b/src/javatest/com/cloudera/flume/conf/TestFlumeBuilder.java
index f578d78..cc2c15e 100644
--- a/src/javatest/com/cloudera/flume/conf/TestFlumeBuilder.java
+++ b/src/javatest/com/cloudera/flume/conf/TestFlumeBuilder.java
@@ -30,6 +30,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.cloudera.flume.ExampleData;
+import com.cloudera.flume.collector.CollectorSink;
 import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.handlers.rolling.RollSink;
 
@@ -254,4 +255,29 @@ public class TestFlumeBuilder implements ExampleData {
     FlumeBuilder.getDecoratorNames().contains("regex");
     FlumeBuilder.getSourceNames().contains("collectorSource");
   }
+
+  @Test
+  public void testGenCollector() throws FlumeSpecException {
+    EventSink snk = FlumeBuilder.buildSink(LogicalNodeContext.testingContext(),
+        "collector() { customDfsSink(\"file:///tmp/foo\",\"foo\") }");
+    assertTrue(snk instanceof CollectorSink);
+  }
+
+  @Test
+  public void testGenCollectorArgs() throws FlumeSpecException {
+    EventSink snk = FlumeBuilder
+        .buildSink(LogicalNodeContext.testingContext(),
+            "collector(foo=\"bar\") { customDfsSink(\"file:///tmp/foo\",\"foo\") }");
+    assertTrue(snk instanceof CollectorSink);
+
+    snk = FlumeBuilder
+        .buildSink(LogicalNodeContext.testingContext(),
+            "collector(3,foo=\"bar\") { customDfsSink(\"file:///tmp/foo\",\"foo\") }");
+    assertTrue(snk instanceof CollectorSink);
+
+    snk = FlumeBuilder.buildSink(LogicalNodeContext.testingContext(),
+        "collector(3) { customDfsSink(\"file:///tmp/foo\",\"foo\") }");
+    assertTrue(snk instanceof CollectorSink);
+
+  }
 }
diff --git a/src/javatest/com/cloudera/flume/conf/TestFlumeSpecGen.java b/src/javatest/com/cloudera/flume/conf/TestFlumeSpecGen.java
index 208c64c..46ad3d6 100644
--- a/src/javatest/com/cloudera/flume/conf/TestFlumeSpecGen.java
+++ b/src/javatest/com/cloudera/flume/conf/TestFlumeSpecGen.java
@@ -148,6 +148,18 @@ public class TestFlumeSpecGen {
     assertEquals(orig, out);
   }
 
+  /**
+   * This tests to make sure that a generator sink is parsed properly and then
+   * regenerated by the spec generator.
+   */
+  @Test
+  public void testGen() throws RecognitionException, FlumeSpecException {
+    String orig = "collector( 12345 ) { counter( \"foo\" ) }";
+    CommonTree tree = FlumeBuilder.parseSink(orig);
+    String out = FlumeSpecGen.genEventSink(tree);
+    assertEquals(orig, out);
+  }
+
   @Test
   public void testKeywordArgGen() throws RecognitionException,
       FlumeSpecException {
diff --git a/src/javatest/com/cloudera/flume/conf/TestParser.java b/src/javatest/com/cloudera/flume/conf/TestParser.java
index f408d37..4d9c135 100644
--- a/src/javatest/com/cloudera/flume/conf/TestParser.java
+++ b/src/javatest/com/cloudera/flume/conf/TestParser.java
@@ -227,6 +227,16 @@ public class TestParser implements ExampleData {
   }
 
   @Test
+  public void testGenSinks() throws RecognitionException {
+    String s2 = "collector() { null }";
+    Object o2 = FlumeBuilder.parseSink(s2);
+    LOG.info(s2);
+    LOG.info(toTree(o2));
+    assertEquals("(GEN collector (DECO (SINK null)))", toTree(o2));
+
+  }
+
+  @Test
   public void testBuilder() throws RecognitionException {
     LOG.info("== nodes ==");
 
@@ -326,7 +336,6 @@ public class TestParser implements ExampleData {
     assertEquals("(ROLL (DECO (SINK null)) (DEC 100))", toTree(o));
   }
 
-
   @Test
   public void testCombo() throws RecognitionException {
     LOG.info("== Combo ==");
@@ -427,4 +436,16 @@ public class TestParser implements ExampleData {
 
   }
 
+  @Test
+  public void testGenCollector() throws FlumeSpecException,
+      RecognitionException {
+    String s = "collector() { customDfsSink(\"file:///tmp/foo\",\"foo\") }";
+    Object o = FlumeBuilder.parseSink(s);
+    LOG.info(toTree(o));
+    assertEquals(
+        "(GEN collector (DECO (SINK customDfsSink (STRING \"file:///tmp/foo\") (STRING \"foo\"))))",
+        toTree(o));
+
+  }
+
 }
-- 
1.7.0.4

