Sink

The purpose of a Sink to extract Events from the Channel andforward them to the next Flume Agent in the flow or store them in an externalrepository. A Sink is associated with exactly one Channels, asconfigured in the Flume properties file. There’s one SinkRunner instanceassociated with every configured Sink, and when the Flume framework callsSinkRunner.start(), a new thread is created to drive the Sink (usingSinkRunner.PollingRunner as the thread’s Runnable). This thread managesthe Sink’s lifecycle. The Sink needs to implement the start() andstop() methods that are part of the LifecycleAware interface. TheSink.start() method should initialize the Sink and bring it to a statewhere it can forward the Events to its next destination. TheSink.process() method should do the core processing of extracting theEvent from the Channel and forwarding it. The Sink.stop() methodshould do the necessary cleanup (e.g. releasing resources). The Sinkimplementation also needs to implement the Configurable interface forprocessing its own configuration settings. For example:

public class MySink extends AbstractSink implements Configurable {
  private String myProp;
 
  @Override
  public void configure(Context context) {
    String myProp = context.getString("myProp", "defaultValue");
 
    // Process the myProp value (e.g. validation)
 
    // Store myProp for later retrieval by process() method
    this.myProp = myProp;
  }
 
  @Override
  public void start() {
    // Initialize the connection to the external repository (e.g. HDFS) that
    // this Sink will forward Events to ..
  }
 
  @Override
  public void stop () {
    // Disconnect from the external respository and do any
    // additional cleanup (e.g. releasing resources or nulling-out
    // field values) ..
  }
 
  @Override
  public Status process() throws EventDeliveryException {
    Status status = null;
 
    // Start transaction
    Channel ch = getChannel();
    Transaction txn = ch.getTransaction();
    txn.begin();
    try {
      // This try clause includes whatever Channel operations you want to do
 
      Event event = ch.take();
 
      // Send the Event to the external repository.
      // storeSomeData(e);
 
      txn.commit();
      status = Status.READY;
    } catch (Throwable t) {
      txn.rollback();
 
      // Log exception, handle individual exceptions as needed
 
      status = Status.BACKOFF;
 
      // re-throw all Errors
      if (t instanceof Error) {
        throw (Error)t;
      }
    }
    return status;
  }
}