JavaInputDStream<ConsumerRecord<String, Object>> directKafkaStream = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent(),
ConsumerStrategies.Subscribe(topics, kafkaParams));
JavaInputDStream<ConsumerRecord<String, Object>> directKafkaStream = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent(),
ConsumerStrategies.Subscribe(topics, kafkaParams));