### Question:
OutputKafka extends Output<LogFeederProps, InputFileMarker> { @Override public void init(LogFeederProps logFeederProps) throws Exception { this.logFeederProps = logFeederProps; Properties props = initProperties(); producer = creteKafkaProducer(props); createKafkaRetryThread(); }  @Override String getStatMetricName(); @Override String getWriteBytesMetricName(); @Override void init(LogFeederProps logFeederProps); @Override synchronized void write(String block, InputFileMarker inputMarker); @Override void setDrain(boolean drain); void flush(); @Override void close(); @Override String getShortDescription(); @Override void copyFile(File inputFile, InputMarker inputMarker); @Override String getOutputType(); @Override Long getPendingCount();  }### Answer:
@Test public void testOutputKafka_noBrokerList() throws Exception { 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(new LogFeederProps()); }
@Test public void testOutputKafka_noTopic() throws Exception { 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(new LogFeederProps()); }
@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(new LogFeederProps()); }
@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(new LogFeederProps()); }