OutputKafka extends Output { @Override public void init() throws Exception { super.init(); Properties props = initProperties(); producer = creteKafkaProducer(props); createKafkaRetryThread(); }  @Override void init(); @Override synchronized void write(String block, InputMarker inputMarker); @Override void setDrain(boolean drain); void flush(); @Override void close(); @Override String getShortDescription(); @Override void copyFile(File inputFile, InputMarker inputMarker);  }
@Test public void testOutputKafka_noBrokerList() throws Exception { LOG.info("testOutputKafka_noBrokerList()"); expectedException.expect(Exception.class); expectedException.expectMessage("For kafka output, bootstrap broker_list is needed"); Map<String, Object> config = new HashMap<String, Object>(); config.put("topic", TEST_TOPIC); outputKafka.loadConfig(config); outputKafka.init(); }
@Test public void testOutputKafka_noTopic() throws Exception { LOG.info("testOutputKafka_noBrokerList()"); expectedException.expect(Exception.class); expectedException.expectMessage("For kafka output, topic is needed"); Map<String, Object> config = new HashMap<String, Object>(); config.put("broker_list", "some broker list"); outputKafka.loadConfig(config); outputKafka.init(); }