spark-streaming-load-kafka-offset.scala

val storedOffsets: Option[mutable.Map[TopicPartition, Long]] = loadOffsets(spark, kuduContext)

val kafkaDStream = storedOffsets match {
    case None =>
        LOGGER.info("storedOffsets was None")
        kafkaParams += ("auto.offset.reset" -> "latest")
        KafkaUtils.createDirectStream[String, Array[Byte]]
            (ssc, PreferConsistent, ConsumerStrategies.Subscribe[String, Array[Byte]]
                (topicsSet, kafkaParams)
        )
    case Some(fromOffsets) =>
        LOGGER.info("storedOffsets was Some(" + fromOffsets + ")")
        kafkaParams += ("auto.offset.reset" -> "none")
        KafkaUtils.createDirectStream[String, Array[Byte]]
            (ssc, PreferConsistent, ConsumerStrategies.Assign[String, Array[Byte]]
                (fromOffsets.keys.toList, kafkaParams, fromOffsets)
        )
}