DefaultSinkFactory implements SinkFactory { @Override public Sink create(String name, String type) throws FlumeException { Preconditions.checkNotNull(name, "name"); Preconditions.checkNotNull(type, "type"); logger.info("Creating instance of sink: {}, type: {}", name, type); Class<? extends Sink> sinkClass = getClass(type); try { Sink sink = sinkClass.newInstance(); sink.setName(name); return sink; } catch (Exception ex) { throw new FlumeException("Unable to create sink: " + name + ", type: " + type + ", class: " + sinkClass.getName(), ex); } }  @Override Sink create(String name, String type); @SuppressWarnings("unchecked") @Override Class<? extends Sink> getClass(String type);  }
@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.assertNotSame(avroSink1, s1); Assert.assertNotSame(avroSink2, s2); }