KafkaLoader extends Loader<LinkConfiguration,ToJobConfiguration> { @Override public void load(LoaderContext context,LinkConfiguration linkConfiguration, ToJobConfiguration jobConfiguration) throws Exception { producer = getProducer(linkConfiguration); System.out.println("got producer"); String topic = jobConfiguration.toJobConfig.topic; System.out.println("topic is:"+topic); String batchUUID = UUID.randomUUID().toString(); String record; while ((record = context.getDataReader().readTextRecord()) != null) { KeyedMessage<String, String> data = new KeyedMessage<String, String> (topic, null, batchUUID, record); messageList.add(data); if (messageList.size() >= KafkaConstants.DEFAULT_BATCH_SIZE) { sendToKafka(messageList); } rowsWritten ++; } if (messageList.size() > 0) { sendToKafka(messageList); } producer.close(); }  @Override void load(LoaderContext context,LinkConfiguration linkConfiguration, ToJobConfiguration jobConfiguration); @Override long getRowsWritten();  }
@Test public void testLoader() throws Exception { LoaderContext context = new LoaderContext(null, new DataReader() { private long index = 0L; @Override public Object[] readArrayRecord() { return null; } @Override public String readTextRecord() { if (index++ < NUMBER_OF_ROWS) { return index + "," + (double)index + ",'" + index + "'"; } else { return null; } } @Override public Object readContent() { return null; } }, null); LinkConfiguration linkConf = new LinkConfiguration(); ToJobConfiguration jobConf = new ToJobConfiguration(); linkConf.linkConfig.brokerList = testUtil.getKafkaServerUrl(); linkConf.linkConfig.zookeeperConnect = testUtil.getZkUrl(); jobConf.toJobConfig.topic = TOPIC; loader.load(context, linkConf, jobConf); for(int i=1;i<=NUMBER_OF_ROWS;i++) { MessageAndMetadata<byte[],byte[]> fetchedMsg = testUtil.getNextMessageFromConsumer(TOPIC); Assert.assertEquals(i + "," + (double) i + "," + "'" + i + "'", new String((byte[]) fetchedMsg.message(), "UTF-8")); } }