From f0d43246eed527401ecaf46f88801f0bb06dda25 Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Sat, 21 May 2011 21:04:16 -0700
Subject: [PATCH] FLUME-632: Support arguments in output format configuration

---
 .../java/com/cloudera/flume/conf/FlumeBuilder.java |   31 ++++++++++++++++++++
 .../flume/handlers/debug/ConsoleEventSink.java     |   19 +++++++++++-
 .../com/cloudera/flume/conf/TestFlumeBuilder.java  |    7 ++++
 3 files changed, 55 insertions(+), 2 deletions(-)

diff --git a/flume-core/src/main/java/com/cloudera/flume/conf/FlumeBuilder.java b/flume-core/src/main/java/com/cloudera/flume/conf/FlumeBuilder.java
index 3e97884..717dc0e 100644
--- a/flume-core/src/main/java/com/cloudera/flume/conf/FlumeBuilder.java
+++ b/flume-core/src/main/java/com/cloudera/flume/conf/FlumeBuilder.java
@@ -89,6 +89,7 @@ public class FlumeBuilder {
 
   enum ASTNODE {
     DEC, HEX, OCT, STRING, BOOL, FLOAT, // literals
+    FUN, // function node
     SINK, SOURCE, // sink or source
     KWARG, // kwarg support
     MULTI, DECO, BACKUP, LET, ROLL, GEN, FAILCHAIN, // compound sinks
@@ -359,6 +360,15 @@ public class FlumeBuilder {
       Preconditions.checkArgument(str.startsWith("\"") && str.endsWith("\""));
       str = str.substring(1, str.length() - 1);
       return StringEscapeUtils.unescapeJava(str);
+    case FUN:
+      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:
@@ -367,6 +377,27 @@ public class FlumeBuilder {
     }
   }
 
+  /**
+   * 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 static Pair<String, CommonTree> buildKWArg(CommonTree t) {
     ASTNODE type = ASTNODE.valueOf(t.getText()); // convert to enum
     Preconditions.checkArgument("KWARG".equals(type.toString()));
diff --git a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/ConsoleEventSink.java b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/ConsoleEventSink.java
index e5972ee..fffcffe 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/ConsoleEventSink.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/ConsoleEventSink.java
@@ -23,6 +23,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.FlumeBuilder.FunctionSpec;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
 import com.cloudera.flume.core.Event;
@@ -69,14 +70,28 @@ public class ConsoleEventSink extends EventSink.Base {
     return new SinkBuilder() {
 
       @Override
+      @Deprecated
       public EventSink build(Context context, String... argv) {
+        // updated interface calls build(Context,Object...) instead
+        throw new RuntimeException(
+            "Old sink builder for Console sink should not be exercised");
+      }
+
+      @Override
+      public EventSink create(Context context, Object... argv) {
         Preconditions.checkArgument(argv.length <= 1,
             "usage: console[(format)]");
         OutputFormat fmt = DebugOutputFormat.builder().build();
         if (argv.length >= 1) {
-          // TODO (jon) handle formats with arguments. Requires language update.
           try {
-            fmt = FormatFactory.get().getOutputFormat(argv[0]);
+            if (argv[0] instanceof FunctionSpec) {
+              // special case for output formats specified by functions
+              FunctionSpec fs = (FunctionSpec) argv[0];
+              fmt = FormatFactory.get().createOutputFormat(fs.getName(),
+                  fs.getArgs());
+            } else {
+              fmt = FormatFactory.get().getOutputFormat(argv[0].toString());
+            }
           } catch (FlumeSpecException e) {
             LOG.error("Bad output format name " + argv[0], e);
             throw new IllegalArgumentException("Bad output format name "
diff --git a/flume-core/src/test/java/com/cloudera/flume/conf/TestFlumeBuilder.java b/flume-core/src/test/java/com/cloudera/flume/conf/TestFlumeBuilder.java
index cc2c15e..2da0ea2 100644
--- a/flume-core/src/test/java/com/cloudera/flume/conf/TestFlumeBuilder.java
+++ b/flume-core/src/test/java/com/cloudera/flume/conf/TestFlumeBuilder.java
@@ -17,6 +17,7 @@
  */
 package com.cloudera.flume.conf;
 
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -278,6 +279,12 @@ public class TestFlumeBuilder implements ExampleData {
     snk = FlumeBuilder.buildSink(LogicalNodeContext.testingContext(),
         "collector(3) { customDfsSink(\"file:///tmp/foo\",\"foo\") }");
     assertTrue(snk instanceof CollectorSink);
+  }
 
+  @Test
+  public void testConsoleFuncArg() throws FlumeSpecException {
+    EventSink snk = FlumeBuilder.buildSink(LogicalNodeContext.testingContext(),
+        "console(seqfile(\"BZip2Codec\"))");
+    assertNotNull(snk);
   }
 }
-- 
1.7.1

