// Path: src/main/java/net/pixelcop/sewer/Event.java
// public interface Event extends Writable {
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/Sink.java
// public abstract class Sink extends Plumbing {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(Sink.class);
// 
//   protected Sink subSink = null;
// 
//   /**
//    * Append an Event to the Sink
//    * @param event
//    * @throws IOException
//    */
//   public abstract void append(Event event) throws IOException;
// 
//   public void setSubSink(Sink subSink) {
//     this.subSink = subSink;
//   }
// 
//   public Sink getSubSink() {
//     return subSink;
//   }
// 
//   /**
//    * Create the SubSink, if possible
//    *
//    * @return True if a SubSink was created, false otherwise.
//    */
//   public boolean createSubSink() {
//     if (getSinkFactory() == null) {
//       return false;
//     }
// 
//     subSink = getSinkFactory().build();
//     return true;
//   }
// 
//   @Override
//   protected void finalize() throws Throwable {
//     if (getStatus() != CLOSED) {
//       try {
//         close();
//       } catch (Throwable t) {
//         LOG.warn("Caught during finalizer close(): " + t.getMessage(), t);
//         // rethrow? this should be good enough since the object was disposed of anyway
//       }
//     }
//   }
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/sink/SequenceFileSink.java
// @DrainSink
// public class SequenceFileSink extends BucketedSink {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(SequenceFileSink.class);
// 
//   private static final VLongWritable ONE = new VLongWritable(1L);
// 
//   /**
//    * Configured path to write to
//    */
//   protected String configPath;
// 
//   /**
//    * Reference to {@link Path} object
//    */
//   protected Path dstPath;
// 
//   protected Writer writer;
// 
//   public SequenceFileSink(String[] args) {
//     this.configPath = args[0];
//   }
// 
//   @Override
//   public void close() throws IOException {
//     if (LOG.isDebugEnabled()) {
//       LOG.debug("close() called; currently: " + getStatusString());
//       LOG.debug("Closing: " + HdfsUtil.pathToString(dstPath));
//     }
// 
//     if (writer != null) {
//       writer.close();
//     }
//     nextBucket = null;
//     setStatus(CLOSED);
// 
//     if (LOG.isInfoEnabled()) {
//       LOG.info("Closed: " + HdfsUtil.pathToString(dstPath));
//     }
//   }
// 
//   @Override
//   public void open() throws IOException {
//     LOG.debug("open");
//     setStatus(OPENING);
//     if (nextBucket == null) {
//       generateNextBucket();
//     }
//     createWriter();
//     setStatus(FLOWING);
//   }
// 
//   protected void createWriter() throws IOException {
// 
//     Configuration conf = Node.getInstance().getConf();
// 
//     CompressionCodec codec = HdfsUtil.createCodec();
//     dstPath = new Path(nextBucket + ".seq");
// 
//     writer = SequenceFile.createWriter(
//         conf,
//         Writer.file(dstPath),
//         Writer.keyClass(Node.getInstance().getSource().getEventClass()),
//         Writer.valueClass(VLongWritable.class),
//         Writer.compression(CompressionType.BLOCK, codec)
//         );
// 
//     if (LOG.isInfoEnabled()) {
//       LOG.info("Opened: " + HdfsUtil.pathToString(dstPath));
//     }
// 
//     nextBucket = null;
//   }
// 
//   @Override
//   public String getFileExt() {
//     return ".seq";
//   }
// 
//   @Override
//   public String generateNextBucket() {
//     nextBucket = BucketPath.escapeString(configPath);
//     return nextBucket;
//   }
// 
//   @Override
//   public void append(Event event) throws IOException {
//     writer.append(event, ONE);
//   }
// 
// }


import java.io.IOException;
import net.pixelcop.sewer.Event;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.sink.BucketedSink;
import net.pixelcop.sewer.sink.SequenceFileSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
    String nextBucket = null;
    if (subSink instanceof BucketedSink) {
      nextBucket = ((BucketedSink) subSink).generateNextBucket();
    }
    this.tx = TransactionManager.getInstance().startTx(nextBucket);
    String durablePath = tx.createTxPath(false);
    this.durableSink = new SequenceFileSink(new String[] { durablePath });

    try {
      this.durableSink.open();
    } catch (IOException e) {
      LOG.error("Error opening durable sink at path " + durablePath, e);
      throw e;
    }

    setStatus(FLOWING);
    LOG.debug("flowing");
  }

  /**
   * This appends the event to two separate sinks: the subsink & and our local disk buffer.
   * Writing to the disk buffer happens asynchronously in the background to allow execution
   * to continue on this thread.
   */
  @Override
  public void append(Event event) throws IOException {

    durableSink.append(event);

    if (subSinkError) {
// Path: src/main/java/net/pixelcop/sewer/Event.java
// public interface Event extends Writable {
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/Sink.java
// public abstract class Sink extends Plumbing {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(Sink.class);
// 
//   protected Sink subSink = null;
// 
//   /**
//    * Append an Event to the Sink
//    * @param event
//    * @throws IOException
//    */
//   public abstract void append(Event event) throws IOException;
// 
//   public void setSubSink(Sink subSink) {
//     this.subSink = subSink;
//   }
// 
//   public Sink getSubSink() {
//     return subSink;
//   }
// 
//   /**
//    * Create the SubSink, if possible
//    *
//    * @return True if a SubSink was created, false otherwise.
//    */
//   public boolean createSubSink() {
//     if (getSinkFactory() == null) {
//       return false;
//     }
// 
//     subSink = getSinkFactory().build();
//     return true;
//   }
// 
//   @Override
//   protected void finalize() throws Throwable {
//     if (getStatus() != CLOSED) {
//       try {
//         close();
//       } catch (Throwable t) {
//         LOG.warn("Caught during finalizer close(): " + t.getMessage(), t);
//         // rethrow? this should be good enough since the object was disposed of anyway
//       }
//     }
//   }
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/sink/SequenceFileSink.java
// @DrainSink
// public class SequenceFileSink extends BucketedSink {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(SequenceFileSink.class);
// 
//   private static final VLongWritable ONE = new VLongWritable(1L);
// 
//   /**
//    * Configured path to write to
//    */
//   protected String configPath;
// 
//   /**
//    * Reference to {@link Path} object
//    */
//   protected Path dstPath;
// 
//   protected Writer writer;
// 
//   public SequenceFileSink(String[] args) {
//     this.configPath = args[0];
//   }
// 
//   @Override
//   public void close() throws IOException {
//     if (LOG.isDebugEnabled()) {
//       LOG.debug("close() called; currently: " + getStatusString());
//       LOG.debug("Closing: " + HdfsUtil.pathToString(dstPath));
//     }
// 
//     if (writer != null) {
//       writer.close();
//     }
//     nextBucket = null;
//     setStatus(CLOSED);
// 
//     if (LOG.isInfoEnabled()) {
//       LOG.info("Closed: " + HdfsUtil.pathToString(dstPath));
//     }
//   }
// 
//   @Override
//   public void open() throws IOException {
//     LOG.debug("open");
//     setStatus(OPENING);
//     if (nextBucket == null) {
//       generateNextBucket();
//     }
//     createWriter();
//     setStatus(FLOWING);
//   }
// 
//   protected void createWriter() throws IOException {
// 
//     Configuration conf = Node.getInstance().getConf();
// 
//     CompressionCodec codec = HdfsUtil.createCodec();
//     dstPath = new Path(nextBucket + ".seq");
// 
//     writer = SequenceFile.createWriter(
//         conf,
//         Writer.file(dstPath),
//         Writer.keyClass(Node.getInstance().getSource().getEventClass()),
//         Writer.valueClass(VLongWritable.class),
//         Writer.compression(CompressionType.BLOCK, codec)
//         );
// 
//     if (LOG.isInfoEnabled()) {
//       LOG.info("Opened: " + HdfsUtil.pathToString(dstPath));
//     }
// 
//     nextBucket = null;
//   }
// 
//   @Override
//   public String getFileExt() {
//     return ".seq";
//   }
// 
//   @Override
//   public String generateNextBucket() {
//     nextBucket = BucketPath.escapeString(configPath);
//     return nextBucket;
//   }
// 
//   @Override
//   public void append(Event event) throws IOException {
//     writer.append(event, ONE);
//   }
// 
// }


// Path: src/main/java/net/pixelcop/sewer/sink/durable/ReliableSink.java
import java.io.IOException;
import net.pixelcop.sewer.Event;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.sink.BucketedSink;
import net.pixelcop.sewer.sink.SequenceFileSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

    String nextBucket = null;
    if (subSink instanceof BucketedSink) {
      nextBucket = ((BucketedSink) subSink).generateNextBucket();
    }
    this.tx = TransactionManager.getInstance().startTx(nextBucket);
    String durablePath = tx.createTxPath(false);
    this.durableSink = new SequenceFileSink(new String[] { durablePath });

    try {
      this.durableSink.open();
    } catch (IOException e) {
      LOG.error("Error opening durable sink at path " + durablePath, e);
      throw e;
    }

    setStatus(FLOWING);
    LOG.debug("flowing");
  }

  /**
   * This appends the event to two separate sinks: the subsink & and our local disk buffer.
   * Writing to the disk buffer happens asynchronously in the background to allow execution
   * to continue on this thread.
   */
  @Override
  public void append(Event event) throws IOException {

    durableSink.append(event);

    if (subSinkError) {
      return;