class SinkProcessorFactory {

  @SuppressWarnings("unchecked")
  public static SinkProcessor getProcessor(Context context, List<Sink> sinks) {
    Preconditions.checkNotNull(context);
    Preconditions.checkNotNull(sinks);
    Preconditions.checkArgument(!sinks.isEmpty());
    Map<String, String> params = context.getParameters();
    SinkProcessor processor;
    String typeStr = params.get(TYPE);
    SinkProcessorType type = SinkProcessorType.OTHER;
    String processorClassName = typeStr;
    try {
      type = SinkProcessorType.valueOf(typeStr.toUpperCase(Locale.ENGLISH));
    } catch (Exception ex) {
      logger.warn("Sink Processor type {} is a custom type", typeStr);
    }

    if (!type.equals(SinkProcessorType.OTHER)) {
      processorClassName = type.getSinkProcessorClassName();
    }

    logger.debug("Creating instance of sink processor type {}, class {}",
            typeStr, processorClassName);
    Class<? extends SinkProcessor> processorClass = null;
    try {
      processorClass = (Class<? extends SinkProcessor>) Class.forName(
          processorClassName);
    } catch (Exception ex) {
      throw new FlumeException("Unable to load sink processor type: " + typeStr
          + ", class: " + type.getSinkProcessorClassName(), ex);
    }
    try {
      processor = processorClass.newInstance();
    } catch (Exception e) {
      throw new FlumeException("Unable to create sink processor, type: " + typeStr
          + ", class: " + processorClassName, e);
    }

    processor.setSinks(sinks);
    Configurables.configure(processor, context);
    return processor;
  }

}