class KafkaSource extends AbstractSource implements EventDrivenSource, Configurable {

    @Override
    public synchronized void start() {

        super.start();
        sourceCounter.start();
        LOGGER.info("Kafka Source started...");

        // make config object
        ConsumerConfig consumerConfig = new ConsumerConfig(this.parameters);
        consumerConnector = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();

        // flume-conf.properties中consumer.sources.s.custom.topic.name的配置项, 和producer.sinks.r.custom.topic.name必须一样
        String topic = (String) this.parameters.get(KafkaFlumeConstans.CUSTOM_TOPIC_KEY_NAME);
        // 每个consumer可以开启多个线程
        String threadCount = (String) this.parameters.get(KafkaFlumeConstans.CUSTOM_CONSUMER_THREAD_COUNT_KEY_NAME);

        topicCountMap.put(topic, new Integer(threadCount));

        // 对于一个指定的topic, 一个consumer可以开启多个线程去获取数据. 这样每一个线程就是一个KafkaStream
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector
                .createMessageStreams(topicCountMap);

        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

        // now launch all the threads 线程池
        this.executorService = Executors.newFixedThreadPool(Integer.parseInt(threadCount));

        // now create an object to consume the messages
        // 每一个KafkaStream都启动一个ConsumerWorker工作线程
        int tNumber = 0;
        for (final KafkaStream stream : streams) {
            this.executorService.submit(new ConsumerWorker(stream, tNumber, sourceCounter));
            tNumber++;
        }
    }

}