From d097a73f0019830fc2048f29ec18829a243d7c4f Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Thu, 2 Jun 2011 05:24:30 -0700
Subject: [PATCH] FLUME-651: Hook in support output format arguments into relevant sinks.
 - console/escapedCustomDfs/customDfs/CollectorSink

---
 .../cloudera/flume/collector/CollectorSink.java    |   42 +++-
 .../java/com/cloudera/flume/conf/FlumeBuilder.java |   24 ++
 .../com/cloudera/flume/conf/SinkFactoryImpl.java   |    7 +-
 .../flume/handlers/debug/ConsoleEventSink.java     |   15 +-
 .../flume/handlers/debug/TextFileSink.java         |   25 ++-
 .../flume/handlers/hdfs/CustomDfsSink.java         |   50 ++--
 .../flume/handlers/hdfs/EscapedCustomDfsSink.java  |   23 ++-
 .../flume/collector/TestCollectorSink.java         |   15 +-
 .../com/cloudera/flume/conf/TestFlumeBuilder.java  |    7 +
 .../flume/handlers/debug/TestTextFileSink.java     |   50 ++++-
 .../flume/handlers/hdfs/TestCustomDfsSink.java     |   18 ++
 .../handlers/hdfs/TestEscapedCustomOutputDfs.java  |  295 +++++---------------
 flume-docs/src/docs/UserGuide/Appendix             |   26 ++-
 flume-docs/src/docs/UserGuide/UsingOutput          |   51 +++-
 14 files changed, 374 insertions(+), 274 deletions(-)

diff --git a/flume-core/src/main/java/com/cloudera/flume/collector/CollectorSink.java b/flume-core/src/main/java/com/cloudera/flume/collector/CollectorSink.java
index c6f9201..1ba5fff 100644
--- a/flume-core/src/main/java/com/cloudera/flume/collector/CollectorSink.java
+++ b/flume-core/src/main/java/com/cloudera/flume/collector/CollectorSink.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
 
 import com.cloudera.flume.agent.FlumeNode;
 import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.FlumeBuilder.FunctionSpec;
 import com.cloudera.flume.conf.FlumeConfiguration;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
@@ -289,21 +290,40 @@ public class CollectorSink extends EventSink.Base {
   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])]");
+      public EventSink create(Context context, Object... argv) {
+        Preconditions.checkArgument(argv.length <= 4 && argv.length >= 2,
+            "usage: collectorSink[(dfsdir,path[,rollmillis], format)]");
         String logdir = FlumeConfiguration.get().getCollectorDfsDir(); // default
         long millis = FlumeConfiguration.get().getCollectorRollMillis();
         String prefix = "";
+        Object format = null;
         if (argv.length >= 1) {
-          logdir = argv[0]; // override
+          logdir = argv[0].toString();
         }
         if (argv.length >= 2) {
-          prefix = argv[1];
+          prefix = argv[1].toString();
         }
         if (argv.length >= 3) {
-          millis = Long.parseLong(argv[2]);
+          // TODO eventually Long instead of String
+          millis = Long.parseLong(argv[2].toString());
+        }
+        if (argv.length >= 4) {
+          // shove format in to context to pass down.
+          format = argv[3];
+          class FormatContext extends Context {
+            FormatContext(Context ctx, Object format) {
+              super(ctx);
+              this.putObj("format", format);
+            }
+          }
+          if (!(format instanceof FunctionSpec)) {
+            LOG.warn("Deprecated syntax: Expected a format spec but instead "
+                + "had a (" + format.getClass().getSimpleName() + ") "
+                + format.toString());
+          }
+          context = new FormatContext(context, format);
         }
+
         try {
           EventSink snk = new CollectorSink(context, logdir, prefix, millis,
               new ProcessTagger(), 250, FlumeNode.getInstance()
@@ -315,6 +335,16 @@ public class CollectorSink extends EventSink.Base {
               "usage: collectorSink[(dfsdir,path[,rollmillis])]" + e);
         }
       }
+
+      @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");
+
+      }
+
     };
   }
 }
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 209d297..b7e6d61 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
@@ -20,6 +20,7 @@ package com.cloudera.flume.conf;
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -42,6 +43,8 @@ import com.cloudera.flume.core.EventSinkDecorator;
 import com.cloudera.flume.core.EventSource;
 import com.cloudera.flume.core.FanOutSink;
 import com.cloudera.flume.handlers.rolling.RollSink;
+import com.cloudera.flume.handlers.text.FormatFactory;
+import com.cloudera.flume.handlers.text.output.OutputFormat;
 import com.cloudera.flume.master.availability.FailoverChainSink;
 import com.cloudera.util.Pair;
 import com.google.common.base.Preconditions;
@@ -399,8 +402,29 @@ public class FlumeBuilder {
     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()));
diff --git a/flume-core/src/main/java/com/cloudera/flume/conf/SinkFactoryImpl.java b/flume-core/src/main/java/com/cloudera/flume/conf/SinkFactoryImpl.java
index 1f9f659..1d52fe2 100644
--- a/flume-core/src/main/java/com/cloudera/flume/conf/SinkFactoryImpl.java
+++ b/flume-core/src/main/java/com/cloudera/flume/conf/SinkFactoryImpl.java
@@ -128,8 +128,11 @@ public class SinkFactoryImpl extends SinkFactory {
       { "text", TextFileSink.builder() },
       { "seqfile", SeqfileEventSink.builder() },
       { "dfs", DFSEventSink.builder() }, // escapes
-      { "customdfs", CustomDfsSink.builder() }, // does not escape
-      { "escapedCustomDfs", EscapedCustomDfsSink.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() },
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..d727922 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,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.FlumeBuilder;
+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 +71,21 @@ 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();
+        OutputFormat fmt = DebugOutputFormat.builder().create();
         if (argv.length >= 1) {
-          // TODO (jon) handle formats with arguments. Requires language update.
           try {
-            fmt = FormatFactory.get().getOutputFormat(argv[0]);
+            fmt = FlumeBuilder.createFormat(FormatFactory.get(), argv[0]);
           } 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/main/java/com/cloudera/flume/handlers/debug/TextFileSink.java b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/TextFileSink.java
index ee1a576..781ca6e 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/TextFileSink.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/TextFileSink.java
@@ -26,7 +26,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeSpecException;
+import com.cloudera.flume.conf.FlumeBuilder.FunctionSpec;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
 import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventSink;
@@ -100,28 +102,43 @@ public class TextFileSink extends EventSink.Base {
 
   public static SinkBuilder builder() {
     return new SinkBuilder() {
+
+      @Deprecated
       @Override
       public EventSink build(Context context, String... args) {
+        // updated interface calls build(Context,Object...) instead
+        throw new RuntimeException(
+            "Old sink builder for Text File sink should not be exercised");
+
+      }
+
+      @Override
+      public EventSink create(Context context, Object... args) {
         Preconditions.checkArgument(args.length >= 1 && args.length <= 2,
             "usage: text(filename[,format])");
         OutputFormat fmt = DebugOutputFormat.builder().build();
-        String val = null;
+        Object val = null;
         if (args.length >= 2) {
           val = args[1];
         } else {
-          val = context.getValue("format");
+          try {
+            val = context.getObj("format", FunctionSpec.class);
+          } catch (ClassCastException cce) {
+            val = context.getObj("format", String.class);
+          }
         }
 
         if (val != null) {
           try {
-            fmt = FormatFactory.get().getOutputFormat(val);
+            fmt = FlumeBuilder.createFormat(FormatFactory.get(), val);
           } catch (FlumeSpecException e) {
             LOG.error("Illegal output format " + args[1], e);
             throw new IllegalArgumentException("Illegal output format" + val);
           }
         }
-        return new TextFileSink(args[0], fmt);
+        return new TextFileSink(args[0].toString(), fmt);
       }
+
     };
   }
 
diff --git a/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java b/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
index faae953..4d5b7e7 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
@@ -30,11 +30,12 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
 import org.apache.hadoop.io.compress.Compressor;
-import org.apache.hadoop.io.compress.GzipCodec;
 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.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeConfiguration;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
@@ -90,13 +91,13 @@ public class CustomDfsSink extends EventSink.Base {
   }
 
   /**
-   * 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.
+   * 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 {
@@ -140,7 +141,8 @@ public class CustomDfsSink extends EventSink.Base {
       return true;
     }
     if (simpleName.endsWith("Codec")) {
-      String prefix = simpleName.substring(0, simpleName.length() - "Codec".length());
+      String prefix = simpleName.substring(0,
+          simpleName.length() - "Codec".length());
       if (prefix.equalsIgnoreCase(codecName)) {
         return true;
       }
@@ -151,7 +153,8 @@ public class CustomDfsSink extends EventSink.Base {
   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)
+    // 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");
@@ -174,10 +177,11 @@ public class CustomDfsSink extends EventSink.Base {
         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);
+    } 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;
   }
@@ -185,22 +189,30 @@ public class CustomDfsSink extends EventSink.Base {
   public static SinkBuilder builder() {
     return new SinkBuilder() {
       @Override
-      public EventSink build(Context context, String... args) {
+      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\")");
+              "usage: customdfs(\"[(hdfs|file|s3n|...)://namenode[:port]]/path\", format)");
         }
 
-        String format = (args.length == 1) ? null : args[1];
+        Object format = (args.length == 1) ? null : args[1];
         OutputFormat fmt;
         try {
-          fmt = FormatFactory.get().getOutputFormat(format);
+          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], fmt);
+        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");
       }
     };
   }
diff --git a/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/EscapedCustomDfsSink.java b/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/EscapedCustomDfsSink.java
index d651f7f..81caaca 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/EscapedCustomDfsSink.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/EscapedCustomDfsSink.java
@@ -27,6 +27,8 @@ 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.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeConfiguration;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
@@ -108,7 +110,7 @@ public class EscapedCustomDfsSink extends EventSink.Base {
    * drawn from the supplied event
    */
   @Override
-  public void append(Event e) throws IOException, InterruptedException  {
+  public void append(Event e) throws IOException, InterruptedException {
     CustomDfsSink w = writer;
     if (shouldSub) {
       String realPath = e.escapeString(absolutePath);
@@ -152,24 +154,24 @@ public class EscapedCustomDfsSink extends EventSink.Base {
   public static SinkBuilder builder() {
     return new SinkBuilder() {
       @Override
-      public EventSink build(Context context, String... args) {
+      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];
+          filename = args[1].toString();
         }
 
-        String format = FlumeConfiguration.get().getDefaultOutputFormat();
+        Object format = FlumeConfiguration.get().getDefaultOutputFormat();
         if (args.length >= 3) {
           format = args[2];
         }
 
         OutputFormat o;
         try {
-          o = FormatFactory.get().getOutputFormat(format);
+          o = FlumeBuilder.createFormat(FormatFactory.get(), format);
         } catch (FlumeSpecException e) {
           LOG.warn("Illegal format type " + format + ".", e);
           o = null;
@@ -177,8 +179,17 @@ public class EscapedCustomDfsSink extends EventSink.Base {
         Preconditions.checkArgument(o != null, "Illegal format type " + format
             + ".");
 
-        return new EscapedCustomDfsSink(args[0], filename, o);
+        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");
       }
+
     };
   }
 }
diff --git a/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java b/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java
index 65c7126..48cf495 100644
--- a/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java
+++ b/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java
@@ -111,12 +111,25 @@ public class TestCollectorSink {
     // 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, 1000)";
+    String src4 = "collectorSink(\"file:///tmp/test\", \"bkjlasdf\", 1000, avro, 1231)";
     FlumeBuilder.buildSink(new Context(), src4);
   }
 
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);
   }
 }
diff --git a/flume-core/src/test/java/com/cloudera/flume/handlers/debug/TestTextFileSink.java b/flume-core/src/test/java/com/cloudera/flume/handlers/debug/TestTextFileSink.java
index c2b37fe..f6704b6 100644
--- a/flume-core/src/test/java/com/cloudera/flume/handlers/debug/TestTextFileSink.java
+++ b/flume-core/src/test/java/com/cloudera/flume/handlers/debug/TestTextFileSink.java
@@ -35,6 +35,7 @@ import com.cloudera.flume.conf.ReportTestingContext;
 import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.handlers.avro.AvroDataFileOutputFormat;
 import com.cloudera.flume.handlers.avro.AvroJsonOutputFormat;
+import com.cloudera.flume.handlers.seqfile.SequenceFileOutputFormat;
 import com.cloudera.flume.handlers.text.output.RawOutputFormat;
 import com.cloudera.flume.reporter.ReportEvent;
 import com.cloudera.flume.reporter.ReportUtil;
@@ -45,7 +46,7 @@ public class TestTextFileSink {
       .getLogger(TestTextFileSink.class);
 
   @Test
-  public void testTextKWArg() throws RecognitionException, FlumeSpecException {
+  public void testTextKWArgDeprecated() throws RecognitionException, FlumeSpecException {
     String s = "text(\"filename\", format=\"raw\")";
     TextFileSink snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
     assertTrue(snk.getFormat() instanceof RawOutputFormat);
@@ -57,6 +58,53 @@ public class TestTextFileSink {
     s = "text(\"filename\", format=\"avrojson\")";
     snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
     assertTrue(snk.getFormat() instanceof AvroJsonOutputFormat);
+
+    s = "text(\"filename\", format=\"seqfile\")";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof SequenceFileOutputFormat);
+  }
+
+  @Test
+  public void testTextKWArg() throws RecognitionException, FlumeSpecException {
+    String s = "text(\"filename\", format=raw)";
+    TextFileSink snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof RawOutputFormat);
+
+    s = "text(\"filename\", format=avrodata)";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof AvroDataFileOutputFormat);
+
+    s = "text(\"filename\", format=avrojson)";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof AvroJsonOutputFormat);
+
+    s = "text(\"filename\", format=seqfile)";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof SequenceFileOutputFormat);
+
+    // this syntax is unsupported in deprecated outptu format args
+    s = "text(\"filename\", format=seqfile(\"bzip2\"))";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof SequenceFileOutputFormat);
+  }
+
+  @Test
+  public void testTextFormatArg() throws RecognitionException, FlumeSpecException {
+    String s = "text(\"filename\", raw)";
+    TextFileSink snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof RawOutputFormat);
+
+    s = "text(\"filename\", avrodata)";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof AvroDataFileOutputFormat);
+
+    s = "text(\"filename\", avrojson)";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof AvroJsonOutputFormat);
+
+    s = "text(\"filename\", seqfile)";
+    snk = (TextFileSink) FlumeBuilder.buildSink(new Context(), s);
+    assertTrue(snk.getFormat() instanceof SequenceFileOutputFormat);
   }
 
   /**
diff --git a/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestCustomDfsSink.java b/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestCustomDfsSink.java
index b761354..b82892b 100644
--- a/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestCustomDfsSink.java
+++ b/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestCustomDfsSink.java
@@ -19,7 +19,10 @@ package com.cloudera.flume.handlers.hdfs;
 
 import static org.junit.Assert.assertEquals;
 
+import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeConfiguration;
+import com.cloudera.flume.conf.FlumeSpecException;
 
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.GzipCodec;
@@ -44,4 +47,19 @@ public class TestCustomDfsSink {
     checkCodec(GzipCodec.class, "org.apache.hadoop.io.compress.GzipCodec");
   }
 
+  @Test
+  public void testOutputFormats() throws FlumeSpecException {
+    // format
+    String src = "formatDfs(\"file:///tmp/test/testfilename\", avro)";
+    FlumeBuilder.buildSink(new Context(), src);
+
+    // format
+    src = "formatDfs(\"file:///tmp/test/testfilename\", seqfile)";
+    FlumeBuilder.buildSink(new Context(), src);
+
+    // format
+    src = "formatDfs(\"file:///tmp/test/testfilename\", seqfile(\"bzip2\"))";
+    FlumeBuilder.buildSink(new Context(), src);
+  }
+
 }
diff --git a/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestEscapedCustomOutputDfs.java b/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestEscapedCustomOutputDfs.java
index 471bdfd..7618acd 100644
--- a/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestEscapedCustomOutputDfs.java
+++ b/flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestEscapedCustomOutputDfs.java
@@ -30,13 +30,17 @@ import java.io.InputStream;
 import java.util.zip.GZIPInputStream;
 
 import org.apache.hadoop.io.compress.BZip2Codec;
+import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
 import org.apache.hadoop.io.compress.DefaultCodec;
+import org.apache.hadoop.io.compress.GzipCodec;
 import org.apache.log4j.Logger;
 import org.junit.Test;
 
 import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeConfiguration;
+import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
 import com.cloudera.flume.core.Attributes;
 import com.cloudera.flume.core.Event;
@@ -45,6 +49,7 @@ import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.handlers.avro.AvroJsonOutputFormat;
 import com.cloudera.flume.handlers.text.SyslogEntryFormat;
 import com.cloudera.flume.handlers.text.output.Log4jOutputFormat;
+import com.cloudera.flume.handlers.text.output.OutputFormat;
 import com.cloudera.util.FileUtil;
 
 /**
@@ -56,16 +61,22 @@ public class TestEscapedCustomOutputDfs {
   final public static Logger LOG = Logger
       .getLogger(TestEscapedCustomOutputDfs.class);
 
-  @Test
-  public void testAvroOutputFormat() throws IOException, InterruptedException {
+  void checkOutputFormat(String format, OutputFormat of) throws IOException,
+      InterruptedException {
+    checkOutputFormat(format, of, "None", null);
+  }
+
+  void checkOutputFormat(String format, OutputFormat of, String codecName,
+      CompressionCodec codec) throws IOException, InterruptedException {
     // set the output format.
     FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "avrojson");
+    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, format);
+    conf.set(FlumeConfiguration.COLLECTOR_DFS_COMPRESS_CODEC, codecName);
 
     // build a sink that outputs to that format.
     File f = FileUtil.mktempdir();
     SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
+    EventSink snk = builder.create(new Context(), "file:///" + f.getPath()
         + "/sub-%{service}");
     Event e = new EventImpl("this is a test message".getBytes());
     Attributes.setString(e, "service", "foo");
@@ -74,180 +85,75 @@ public class TestEscapedCustomOutputDfs {
     snk.close();
 
     ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    AvroJsonOutputFormat ajof = new AvroJsonOutputFormat();
-    ajof.format(exWriter, e);
+    of.format(exWriter, e);
     exWriter.close();
     String expected = new String(exWriter.toByteArray());
 
     // check the output to make sure it is what we expected.
-    File fo = new File(f.getPath() + "/sub-foo");
 
-    FileReader fr = new FileReader(fo);
-    BufferedReader br = new BufferedReader(fr);
-    String read = br.readLine() + "\n";
-    assertEquals(expected, read);
+    // handle compression codec / extensions when checking.
+    String ext = ""; // file extension
+    if (codec != null) {
+      ext = codec.getDefaultExtension();
+    }
+    InputStream in = new FileInputStream(f.getPath() + "/sub-foo" + ext);
+    if (codec != null) {
+      in = codec.createInputStream(in);
+    }
+    byte[] buf = new byte[1];
+    StringBuilder output = new StringBuilder();
+    // read the file
+    while ((in.read(buf)) > 0) {
+      output.append(new String(buf));
+    }
+    in.close(); // Must close for windows to delete
+    assertEquals(expected, output.toString());
+
+    // This doesn't get deleted in windows but the core test succeeds
+    assertTrue("temp folder successfully deleted", FileUtil.rmr(f));
   }
 
   @Test
-  public void testSyslogOutputFormat() throws IOException, InterruptedException {
-    // set the output format.
-    FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "syslog");
-
-    // build a sink that outputs to that format.
-    File f = FileUtil.mktempdir();
-    SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
-        + "/sub-%{service}");
-    Event e = new EventImpl("this is a test message".getBytes());
-    Attributes.setString(e, "service", "foo");
-    snk.open();
-    snk.append(e);
-    snk.close();
-
-    ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    SyslogEntryFormat fmt = new SyslogEntryFormat();
-    fmt.format(exWriter, e);
-    exWriter.close();
-    String expected = new String(exWriter.toByteArray());
-
-    // check the output to make sure it is what we expected.
-    File fo = new File(f.getPath() + "/sub-foo");
+  public void testAvroOutputFormat() throws IOException, InterruptedException {
+    checkOutputFormat("avrojson", new AvroJsonOutputFormat());
+  }
 
-    FileReader fr = new FileReader(fo);
-    BufferedReader br = new BufferedReader(fr);
-    String read = br.readLine() + "\n";
-    assertEquals(expected, read);
+  @Test
+  public void testSyslogOutputFormat() throws IOException, InterruptedException {
+    checkOutputFormat("syslog", new SyslogEntryFormat());
   }
 
   @Test
   public void testLog4jOutputFormat() throws IOException, InterruptedException {
-    // set the output format.
-    FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "log4j");
-
-    // build a sink that outputs to that format.
-    File f = FileUtil.mktempdir();
-    SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
-        + "/sub-%{service}");
-    Event e = new EventImpl("this is a test message".getBytes());
-    Attributes.setString(e, "service", "foo");
-    snk.open();
-    snk.append(e);
-    snk.close();
-
-    ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    Log4jOutputFormat fmt = new Log4jOutputFormat();
-    fmt.format(exWriter, e);
-    exWriter.close();
-    String expected = new String(exWriter.toByteArray());
-
-    // check the output to make sure it is what we expected.
-    File fo = new File(f.getPath() + "/sub-foo");
-    FileReader fr = new FileReader(fo);
-    BufferedReader br = new BufferedReader(fr);
-    String read = br.readLine() + "\n";
-    assertEquals(expected, read);
+    checkOutputFormat("log4j", new Log4jOutputFormat());
   }
 
   /**
    * Test to write few log lines, compress using gzip, write to disk, read back
    * the compressed file and verify the written lines. This test alone doesn't
-   * test GZipCodec with its Native Libs. java.library.path must contain the path to the
-   * hadoop native libs for this to happen.
-   * 
+   * test GZipCodec with its Native Libs. java.library.path must contain the
+   * path to the hadoop native libs for this to happen.
+   *
    * @throws IOException
    * @throws InterruptedException
    */
   @Test
   public void testGZipCodec() throws IOException, InterruptedException {
-    // set the output format.
-    FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "syslog");
-    conf.set(FlumeConfiguration.COLLECTOR_DFS_COMPRESS_CODEC, "GzipCodec");
-
-    // build a sink that outputs to that format.
-    File f = FileUtil.mktempdir();
-    SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
-        + "/sub-%{service}");
-    Event e = new EventImpl("this is a test message".getBytes());
-    Attributes.setString(e, "service", "foo");
-    snk.open();
-    snk.append(e);
-    snk.close();
-
-    ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    SyslogEntryFormat fmt = new SyslogEntryFormat();
-    fmt.format(exWriter, e);
-    exWriter.close();
-    String expected = new String(exWriter.toByteArray());
-
-    // check the output to make sure it is what we expected.
-    // read the gzip file and verify the contents
-
-    GZIPInputStream gzin = new GZIPInputStream(new FileInputStream(f.getPath()
-        + "/sub-foo.gz"));
-    byte[] buf = new byte[1];
-    StringBuilder output = new StringBuilder();
-
-    while ((gzin.read(buf)) > 0) {
-      output.append(new String(buf));
-    }
-    gzin.close(); // Must close for windows to delete
-    assertEquals(expected, output.toString());
-
-    // This doesn't get deleted in windows but the core test succeeds
-    assertTrue("temp folder successfully deleted", FileUtil.rmr(f));
+    checkOutputFormat("syslog", new SyslogEntryFormat(), "GzipCodec",
+        new GzipCodec());
   }
 
   /**
    * Test to write few log lines, compress using bzip2, write to disk, read back
    * the compressed file and verify the written lines.
-   * 
+   *
    * @throws IOException
    * @throws InterruptedException
    */
   @Test
   public void testBZip2Codec() throws IOException, InterruptedException {
-    // set the output format.
-    FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "syslog");
-    conf.set(FlumeConfiguration.COLLECTOR_DFS_COMPRESS_CODEC, "BZip2Codec");
-
-    // build a sink that outputs to that format.
-    File f = FileUtil.mktempdir();
-    SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
-        + "/sub-%{service}");
-    Event e = new EventImpl("this is a test message".getBytes());
-    Attributes.setString(e, "service", "foo");
-    snk.open();
-    snk.append(e);
-    snk.close();
-
-    ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    SyslogEntryFormat fmt = new SyslogEntryFormat();
-    fmt.format(exWriter, e);
-    exWriter.close();
-    String expected = new String(exWriter.toByteArray());
-
-    // check the output to make sure it is what we expected.
-    // read the bzip2 file and verify the contents
-    BZip2Codec bz2Codec = new BZip2Codec();
-    InputStream bz2in = bz2Codec.createInputStream(new FileInputStream(f
-        .getPath() + "/sub-foo.bz2"));
-    byte[] buf = new byte[1];
-    StringBuilder output = new StringBuilder();
-
-    while ((bz2in.read(buf)) > 0) {
-      output.append(new String(buf));
-    }
-    bz2in.close(); // Must close for windows to delete
-    assertEquals(expected, output.toString());
-
-    assertTrue("temp folder successfully deleted", FileUtil.rmr(f));
+    checkOutputFormat("syslog", new SyslogEntryFormat(), "BZip2Codec",
+        new BZip2Codec());
   }
 
   /**
@@ -260,97 +166,42 @@ public class TestEscapedCustomOutputDfs {
    * @throws InterruptedException
    */
   @Test
-  public void testBZip2CodecWrongCase() throws IOException, InterruptedException {
-    // set the output format.
-    FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "syslog");
-    conf.set(FlumeConfiguration.COLLECTOR_DFS_COMPRESS_CODEC, "bzip2Codec");
-
-    // build a sink that outputs to that format.
-    File f = FileUtil.mktempdir();
-    SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
-        + "/sub-%{service}");
-    Event e = new EventImpl("this is a test message".getBytes());
-    Attributes.setString(e, "service", "foo");
-    snk.open();
-    snk.append(e);
-    snk.close();
-
-    ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    SyslogEntryFormat fmt = new SyslogEntryFormat();
-    fmt.format(exWriter, e);
-    exWriter.close();
-    String expected = new String(exWriter.toByteArray());
-
-    // check the output to make sure it is what we expected.
-    // read the bzip2 file and verify the contents
-    BZip2Codec bz2Codec = new BZip2Codec();
-    InputStream bz2in = bz2Codec.createInputStream(new FileInputStream(f
-        .getPath() + "/sub-foo.bz2"));
-    byte[] buf = new byte[1];
-    StringBuilder output = new StringBuilder();
-
-    while ((bz2in.read(buf)) > 0) {
-      output.append(new String(buf));
-    }
-    bz2in.close(); // Must close for windows to delete
-    assertEquals(expected, output.toString());
-
-    assertTrue("temp folder successfully deleted", FileUtil.rmr(f));
+  public void testBZip2CodecWrongCase() throws IOException,
+      InterruptedException {
+    checkOutputFormat("syslog", new SyslogEntryFormat(), "bzip2Codec",
+        new BZip2Codec());
   }
 
   /**
    * Test to write few log lines, compress using default, write to disk, read
    * back the compressed file and verify the written lines.
-   * 
+   *
    * @throws InterruptedException
    */
-
   @Test
   public void testDefaultCodec() throws IOException, InterruptedException {
-    // set the output format.
-    FlumeConfiguration conf = FlumeConfiguration.get();
-    conf.set(FlumeConfiguration.COLLECTOR_OUTPUT_FORMAT, "syslog");
-    conf.set(FlumeConfiguration.COLLECTOR_DFS_COMPRESS_CODEC, "DefaultCodec");
-
-    // build a sink that outputs to that format.
-    File f = FileUtil.mktempdir();
-    SinkBuilder builder = EscapedCustomDfsSink.builder();
-    EventSink snk = builder.build(new Context(), "file:///" + f.getPath()
-        + "/sub-%{service}");
-    Event e = new EventImpl("this is a test message".getBytes());
-    Attributes.setString(e, "service", "foo");
-    snk.open();
-    snk.append(e);
-    snk.close();
-
-    ByteArrayOutputStream exWriter = new ByteArrayOutputStream();
-    SyslogEntryFormat fmt = new SyslogEntryFormat();
-    fmt.format(exWriter, e);
-    exWriter.close();
-    String expected = new String(exWriter.toByteArray());
-
-    // check the output to make sure it is what we expected.
-    // read the file and verify the contents
-    DefaultCodec defaultCodec = new DefaultCodec();
-    defaultCodec.setConf(conf);
-    InputStream defaultIn = defaultCodec.createInputStream(new FileInputStream(
-        f.getPath() + "/sub-foo.deflate"));
-    byte[] buf = new byte[1];
-    StringBuilder output = new StringBuilder();
-
-    while ((defaultIn.read(buf)) > 0) {
-      output.append(new String(buf));
-    }
-    assertEquals(expected, output.toString());
-
-    assertTrue("temp folder successfully deleted", FileUtil.rmr(f));
+    DefaultCodec codec = new DefaultCodec();
+    codec.setConf(FlumeConfiguration.get()); // default needs conf
+    checkOutputFormat("syslog", new SyslogEntryFormat(), "DefaultCodec", codec);
   }
 
   public void testCodecs() {
     LOG.info(CompressionCodecFactory.getCodecClasses(FlumeConfiguration.get()));
+  }
 
+  @Test
+  public void testOutputFormats() throws FlumeSpecException {
+    // format
+    String src = "escapedFormatDfs(\"file:///tmp/test/testfilename\",\"\", avro)";
+    FlumeBuilder.buildSink(new Context(), src);
+
+    // format
+    src = "escapedFormatDfs(\"file:///tmp/test/testfilename\",\"\", seqfile)";
+    FlumeBuilder.buildSink(new Context(), src);
+
+    // format
+    src = "escapedFormatDfs(\"file:///tmp/test/testfilename\",\"\", seqfile(\"bzip2\"))";
+    FlumeBuilder.buildSink(new Context(), src);
   }
 
 }
diff --git a/flume-docs/src/docs/UserGuide/Appendix b/flume-docs/src/docs/UserGuide/Appendix
index b68eca5..c8a8918 100644
--- a/flume-docs/src/docs/UserGuide/Appendix
+++ b/flume-docs/src/docs/UserGuide/Appendix
@@ -65,7 +65,7 @@ configured to listen on TCP port +_port_+. Supports both Apache-Thrift and
 Apache-Avro RPC framework.  The type of RPC framework is specified by +event.rpc.type+ 
 property (THRIFT or AVRO), the default is THRIFT.  Note that same RPC framework is used for rpcSink.
 
-+text("_filename_")+ :: One-time text file source.  One event per +\n+ 
++text("_filename_"[, _format_])+ :: One-time text file source.  One event per +\n+ 
 delimited line.
 
 +tail("_filename_"[, _startFromEnd_=false]{,delim="_regex_",
@@ -188,7 +188,7 @@ uninterpreted arrays of bytes.
 
 [horizontal] 
 
-+collectorSink("_fsdir_","_fsfileprefix_", _rollmillis_)+ :: Collector sink.  
++collectorSink("_fsdir_","_fsfileprefix_"[, _rollmillis_[, _format_]])+ :: Collector sink.  
 +_fsdir_+ is a fs directory URI such as +hdfs://namenode/path+ or +file:///
 path+.  +_fsfileprefix_+ is a file name prefix for outputted files. Both of 
 these can use escape sequences documented to bucket data as documented in the 
@@ -269,10 +269,10 @@ is maintained and automatically selected by the master.
 [horizontal]
 +null+ :: Null sink. Events are dropped
 
-+console[("_formatter_")]+ :: Console sink.  Display events to process's 
++console[(_format_)]+ :: Console sink.  Display events to process's 
 stdout using the optionally specified output formatter.
 
-+text("_txtfile_"[,"_formatter_"])+ :: Textfile sink.  Write to text file 
++text("_txtfile_"[,_format_])+ :: Textfile sink.  Write to text file 
 txtfile, using an optionally specified +_formatter_+.  If a file already 
 exists, this sink will attempt to overwrite it.
 
@@ -286,17 +286,31 @@ in Flume- specific Hadoop seqfile record format. The +_hdfspath_+ can
 use escape sequences to bucket data as documented in the *Output
 Bucketing* section.
 
-+customdfs("_hdfspath_"[, "_format_"])+ :: Hadoop dfs formatted file
++formatDfs("_hdfspath_"[, _format_])+ :: Hadoop dfs formatted file
 sink.  The _hdfspath_ string is *not* escaped. The output format of
 writes to a dfs path in using specified output _format_.
 
-+escapedCustomDfs("_hdfspath_", "_file_", "_format_") :: Hadoop dfs formatted 
++escapedFormatDfs("_hdfspath_", "_file_"[, _format_]) :: Hadoop dfs formatted 
 file sink.  The _hdfspath_ string is escaped and events will get written to 
 particular directories and filenames based on this string. The output format 
 of writes to a dfs path in using specified output _format_. The +_hdfspath_+ 
 can use escape sequences documented to bucket data as documented in the 
 *Output Bucketing* section.
 
++customdfs("_hdfspath_"[, _format_])+ :: Hadoop dfs formatted file
+sink.  The _hdfspath_ string is *not* escaped. The output format of
+writes to a dfs path in using specified output _format_.  This sink is
+being deprecated by +formatDfs+.
+
++escapedCustomDfs("_hdfspath_", "_file_"[, _format_]) :: Hadoop dfs
+formatted file sink.  The _hdfspath_ string is escaped and events will
+get written to particular directories and filenames based on this
+string. The output format of writes to a dfs path in using specified
+output _format_. The +_hdfspath_+ can use escape sequences documented
+to bucket data as documented in the *Output Bucketing* section.  This
+sink is being deprecatd by +escapedFormatDfs+.
+
+
 +rpcSink("_host_"[, _port_])+ :: A remote procedure call (RPC) sink that is 
 configured to send to machine +_host_+ on TCP port +_port_+. Default port is 35861 
 and can be overridden by setting the +flume.collector.event.port+ property. 
diff --git a/flume-docs/src/docs/UserGuide/UsingOutput b/flume-docs/src/docs/UserGuide/UsingOutput
index d2231a2..6ff7e00 100644
--- a/flume-docs/src/docs/UserGuide/UsingOutput
+++ b/flume-docs/src/docs/UserGuide/UsingOutput
@@ -155,10 +155,15 @@ hour with filenames prefixed "web-".
 
 === Output Format
 
-Now that you have control of where files go, this section describes how you can 
-control the output format of data.  Currently, this is set via the 
-'flume.collector.output.format' property set in the flume-site.xml file.  The output 
-formats are:
+Now that you have control of where files go, this section describes
+how you can control the output format of data.  There are two ways to
+this.  The first is to set the default value in flume-site.xml and the other is by specifying output format in particular sinks.  
+
+==== Configuring the default output format.
+
+You can set the default output format by etting the
+'flume.collector.output.format' property set in the +flume-site.xml+
+file.  The output formats are:
 
 .Output formats
 [horizontal]
@@ -171,7 +176,9 @@ log4j :: a log4j pattern similar to that used by CDH output pattern.
 raw :: Event body only.  This is most similar to copying a file but
 does not preserve any uniqifying metadata like host/timestamp/nanos.
 syslog :: a syslog like text output format.
+seqfile :: the binary hadoop Sequence file format with WritableEventKeys keys, and WritableEvent as values.
 
+Here is an example of a property added to the +flume-site.xml+ file:
 ----
 <property>
   <name>flume.collector.output.format</name>
@@ -186,11 +193,47 @@ syslog :: a syslog like text output format.
     log4j - outputs events in a pattern similar to Hadoop's log4j pattern
     raw - Event body only.  This is most similar to copying a file but
       does not preserve any uniqifying metadata like host/timestamp/nanos.
+    seqfile - this is the hadoop sequence file format with
+       WritableEventKeys and WritableEvent objects. 
     syslog - outputs events in a syslog-like format
   </description>
 </property>  
 ----
 
+==== Setting output format in specific sinks.
+
+Some sinks have an optional output format argument.  These include the
++console+, +text+, +customdfs+/+formatDfs+,
++escapedCustomDfs+/+escapedFormatDfs+, and +collectorSink+. 
+
+In these cases, we will be focusing on the optional format argument.
+----
+collectorSink( "dfsdir","prefix"[, rollmillis[, format]])  
+text("file"[,format])
+formatDfs("hdfs://nn/file" [, format])
+escapedFormatDfs("hdfs://nn/file" [, format])
+----
+
+This mechanism is more flexible because it allows for output formats
+with arguments.  Currently, the +seqfile+ output format supports
+arugments for specifying an sequence file internal compression codec.
+Codecs available for use are the same as those available to hadoop --
+generally these include +gzip+, +bzip2+, and can include others that
+have been plugged in such +lzo+ or +snappy+.
+
+The compression codec is specified by adding an argument to output
+format.  For example, the +seqfile+ output format accepts a
+compression codec argument. So, to specify an output format that is a
+sequence file using the bzip2 codec, the output format would be
++seqfile("bzip2")+.  To write bzip2 compressed sequence files, using a
+formatDfs sink, we would specify +formatDfs("hdfs://nn/dir/file",
+seqfile("bzip2"))+
+
+NOTE: The old syntax requires specifying a Flume String (ex:
++console("avrojson")+, +console("seqfile")+ has been deprecated but
+will be supported for a few more versions.  The preferred arugment is
+now a "function". (ex: +console(avrojson)+,
++console(seqfile("bzip2"))+).
 
 === Small Files Compared to High Latency
 
-- 
1.7.1

