@Test public void test() { Context context = new Context(); context.put("type", FailoverSinkProcessor.class.getName()); context.put("priority.sink1", "1"); context.put("priority.sink2", "2"); SinkFactory sf = new DefaultSinkFactory(); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(sf.create("sink1", "avro")); sinks.add(sf.create("sink2", "avro")); SinkProcessor sp = SinkProcessorFactory.getProcessor(context, sinks); context.put("type", "failover"); SinkProcessor sp2 = SinkProcessorFactory.getProcessor(context, sinks); Assert.assertEquals(sp.getClass(), sp2.getClass()); }
@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()); } 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; }
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()); } 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; } }
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()); } 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; }  }
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()); } 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; }  @SuppressWarnings("unchecked") static SinkProcessor getProcessor(Context context,
 List<Sink> sinks); @SuppressWarnings("unchecked") static SinkProcessor getProcessor(ComponentConfiguration conf,
      List<Sink> sinks); }
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()); } 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; }  @SuppressWarnings("unchecked") static SinkProcessor getProcessor(Context context,
 List<Sink> sinks); @SuppressWarnings("unchecked") static SinkProcessor getProcessor(ComponentConfiguration conf,
      List<Sink> sinks);  }