diff --git flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
index 2d1f7e8..42bd58a 100644
--- flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
+++ flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
@@ -102,43 +102,16 @@ public class CustomDfsSink extends EventSink.Base {
     FlumeConfiguration conf = FlumeConfiguration.get();
     FileSystem hdfs;
 
-    String codecName = conf.getCollectorDfsCompressCodec();
-    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 (cls.getSimpleName().equalsIgnoreCase(codecName)) {
-        try {
-          codec = cls.newInstance();
-        } catch (InstantiationException e) {
-          LOG.error("Unable to instantiate " + codec + " class");
-        } catch (IllegalAccessException e) {
-          LOG.error("Unable to access " + codec + " class");
-        }
-      }
-    }
+    CompressionCodec codec = DFSEventSink.getCodec(conf);
 
     if (codec == null) {
-      if (!codecName.equalsIgnoreCase("None")) {
-        LOG.warn("Unsupported compression codec " + codecName
-            + ".  Please choose from: " + codecStrs);
-      }
       dstPath = new Path(path);
       hdfs = dstPath.getFileSystem(conf);
       writer = hdfs.create(dstPath);
       LOG.info("Creating HDFS file: " + dstPath.toString());
       return;
     }
-    //Must check instanceof codec as BZip2Codec doesn't inherit Configurable
-    if(codec instanceof Configurable){
-      //Must set the configuration for Configurable objects that may or do use native libs
-      ((Configurable)codec).setConf(conf);
-    }
+
     Compressor cmp = codec.createCompressor();
     dstPath = new Path(path + codec.getDefaultExtension());
     hdfs = dstPath.getFileSystem(conf);
diff --git flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/DFSEventSink.java flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/DFSEventSink.java
index b780c15..acd04fd 100644
--- flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/DFSEventSink.java
+++ flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/DFSEventSink.java
@@ -18,14 +18,19 @@
 package com.cloudera.flume.handlers.hdfs;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
+import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.SequenceFile.Writer;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,6 +64,43 @@ public class DFSEventSink extends EventSink.Base {
     this.path = path;
     shouldSub = Event.containsTag(path);
   }
+  
+  protected static CompressionCodec getCodec(FlumeConfiguration conf) {
+    String codecName = conf.getCollectorDfsCompressCodec();
+    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 (cls.getSimpleName().equalsIgnoreCase(codecName)) {
+        try {
+          codec = cls.newInstance();
+        } catch (InstantiationException e) {
+          LOG.error("Unable to instantiate " + codec + " class");
+        } catch (IllegalAccessException e) {
+          LOG.error("Unable to access " + codec + " class");
+        }
+      }
+    }
+
+    //Must check instanceof codec as BZip2Codec doesn't inherit Configurable
+    if(codec instanceof Configurable){
+      //Must set the configuration for Configurable objects that may or do use native libs
+      ((Configurable)codec).setConf(conf);
+    }
+    
+    if (codec == null) {
+      if (!codecName.equalsIgnoreCase("None")) {
+        LOG.warn("Unsupported compression codec " + codecName
+            + ".  Please choose from: " + codecStrs);
+      }
+    }
+    return codec;
+  }
 
   protected Writer openWriter(String p) throws IOException {
     LOG.info("Opening " + p);
@@ -67,9 +109,18 @@ public class DFSEventSink extends EventSink.Base {
     Path dstPath = new Path(p);
     FileSystem hdfs = dstPath.getFileSystem(conf);
 
-    Writer w = SequenceFile.createWriter(hdfs, conf, dstPath,
-        WriteableEventKey.class, WriteableEvent.class);
-
+    CompressionCodec codec = getCodec(conf);
+    Writer w;
+    if (codec == null) {
+      System.out.println("Codec is null");
+      w = SequenceFile.createWriter(hdfs, conf, dstPath,
+          WriteableEventKey.class, WriteableEvent.class);
+    } else {
+      System.out.println("Codec is not null");
+      w = SequenceFile.createWriter(hdfs, conf, dstPath,
+          WriteableEventKey.class, WriteableEvent.class,
+          SequenceFile.getCompressionType(conf), codec);
+    }
     return w;
   }
 
diff --git flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestDFSWrite.java flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestDFSWrite.java
index 62f8110..8605503 100644
--- flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestDFSWrite.java
+++ flume-core/src/test/java/com/cloudera/flume/handlers/hdfs/TestDFSWrite.java
@@ -139,6 +139,44 @@ public class TestDFSWrite {
 
     hdfs.close();
   }
+  
+
+  @Test
+  public void testHDFSEventSinkWithCompression() throws IOException,
+      InterruptedException {
+    FlumeConfiguration conf = FlumeConfiguration.createTestableConfiguration();
+    conf.set(FlumeConfiguration.COLLECTOR_DFS_COMPRESS_CODEC, "BZip2Codec");
+    SequenceFile.setCompressionType(conf, SequenceFile.CompressionType.BLOCK);
+    String str = "file:///tmp/testfile";
+    Path path = new Path(str);
+    DFSEventSink sink = new DFSEventSink(str);
+
+    FileSystem hdfs = path.getFileSystem(conf);
+
+    sink.open();
+    Event e = new EventImpl("EVENT".getBytes());
+    sink.append(e);
+    sink.close();
+
+    FileStatus stats = hdfs.getFileStatus(path);
+    assertTrue(stats.getLen() > 0);
+
+    // // reading
+    SequenceFile.Reader r = new SequenceFile.Reader(hdfs, path, conf);
+    assertTrue("Block compressed", r.isBlockCompressed());
+    assertEquals("BZip2Codec",
+        r.getCompressionCodec().getClass().getSimpleName());
+    WriteableEventKey k = new WriteableEventKey();
+    WriteableEvent evt = new WriteableEvent();
+    r.next(k, evt);
+
+    assertEquals(evt.getTimestamp(), e.getTimestamp());
+    assertEquals(evt.getNanos(), e.getNanos());
+    assertEquals(evt.getPriority(), e.getPriority());
+    assertTrue(Arrays.equals(evt.getBody(), e.getBody()));
+
+    hdfs.close();
+  }
 
   /**
    * Test that pathnames are being correctly substituted.
