JavaInputDStream<ConsumerRecord<String, Object>> directKafkaStream =
        KafkaUtils.createDirectStream(ssc,
        LocationStrategies.PreferConsistent(),

        ConsumerStrategies.Subscribe(topics, kafkaParams));

Reply via email to