crawler.java

public void process(String topic, ConsumerConnector consumer) {

        Map<String, Integer> topicCountMap = new HashMap<>();
        topicCountMap.put(topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();

        while (it.hasNext()) {
            try {
                MessageAndMetadata<byte[], byte[]> next = it.next();
                String seedUrl = new String(next.message());
                _logger.info("Processing url [{}]", seedUrl);
                Random random = new Random();
                String batchId = String.valueOf(random.nextInt());
                Configuration nutchConfiguration = NutchConfiguration.create();
                nutchConfiguration.set(BATCH_ID, batchId);
                String solrUrl = nutchConfiguration.get(SOLR_URL);
                String crawlArgs = String.format("-seedurl %s -depth 5 -topN 10", seedUrl);
                // Run Crawl tool
                ToolRunner.run(nutchConfiguration, new Crawler(),
                        tokenize(crawlArgs));
                SolrIndexerJob solrIndexerJob = new SolrIndexerJob();
                solrIndexerJob.setConf(nutchConfiguration);
                solrIndexerJob.indexSolr(solrUrl, batchId);
            } catch (Exception ex) {
                java.util.logging.Logger.getLogger(ABCrawler.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }