KafkaSource extends AbstractSource implements EventDrivenSource, Configurable { @Override public synchronized void start() { super.start(); sourceCounter.start(); LOGGER.info("Kafka Source started..."); ConsumerConfig consumerConfig = new ConsumerConfig(this.parameters); consumerConnector = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig); Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); String topic = (String) this.parameters.get(KafkaFlumeConstans.CUSTOM_TOPIC_KEY_NAME); String threadCount = (String) this.parameters.get(KafkaFlumeConstans.CUSTOM_CONSUMER_THREAD_COUNT_KEY_NAME); topicCountMap.put(topic, new Integer(threadCount)); Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector .createMessageStreams(topicCountMap); List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic); this.executorService = Executors.newFixedThreadPool(Integer.parseInt(threadCount)); int tNumber = 0; for (final KafkaStream stream : streams) { this.executorService.submit(new ConsumerWorker(stream, tNumber, sourceCounter)); tNumber++; } }  @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop();  }
@Test public void testAppend() throws Exception { LOGGER.info("Begin Seding message to Kafka................"); source.start(); sendMessageToKafka(); Thread.sleep(5000); Transaction transaction = channel.getTransaction(); try{ transaction.begin(); Event event; while ((event = channel.take()) != null) { LOGGER.info("#get channel########" + new String(event.getBody(), "UTF-8")); } transaction.commit(); } finally{ transaction.close(); } }