DefaultSinkFactory implements SinkFactory { @SuppressWarnings("unchecked") @Override public Sink create(String name, String type) throws FlumeException { Preconditions.checkNotNull(name); Preconditions.checkNotNull(type); logger.info("Creating instance of sink {} type{}", name, type); String sinkClassName = type; SinkType sinkType = SinkType.OTHER; try { sinkType = SinkType.valueOf(type.toUpperCase()); } catch (IllegalArgumentException ex) { logger.debug("Sink type {} is a custom type", type); } if (!sinkType.equals(SinkType.OTHER)) { sinkClassName = sinkType.getSinkClassName(); } Class<? extends Sink> sinkClass = null; try { sinkClass = (Class<? extends Sink>) Class.forName(sinkClassName); } catch (Exception ex) { throw new FlumeException("Unable to load sink type: " + type + ", class: " + sinkClassName, ex); } Map<String, Sink> sinkMap = sinks.get(sinkClass); if (sinkMap == null) { sinkMap = new HashMap<String, Sink>(); sinks.put(sinkClass, sinkMap); } Sink sink = sinkMap.get(name); if (sink == null) { try { sink = sinkClass.newInstance(); sink.setName(name); sinkMap.put(name, sink); } catch (Exception ex) { sinks.remove(sinkClass); throw new FlumeException("Unable to create sink: " + name + ", type: " + type + ", class: " + sinkClassName, ex); } } return sink; } DefaultSinkFactory(); @Override synchronized boolean unregister(Sink sink); @SuppressWarnings("unchecked") @Override Sink create(String name, String type); synchronized Map<Class<?>, Map<String, Sink>> getRegistryClone();  }
@Test public void testDuplicateCreate() { Sink avroSink1 = sinkFactory.create("avroSink1", "avro"); Sink avroSink2 = sinkFactory.create("avroSink2", "avro"); Assert.assertNotNull(avroSink1); Assert.assertNotNull(avroSink2); Assert.assertNotSame(avroSink1, avroSink2); Assert.assertTrue(avroSink1 instanceof AvroSink); Assert.assertTrue(avroSink2 instanceof AvroSink); Sink s1 = sinkFactory.create("avroSink1", "avro"); Sink s2 = sinkFactory.create("avroSink2", "avro"); Assert.assertSame(avroSink1, s1); Assert.assertSame(avroSink2, s2); }