title: Simple Spark Structured Streaming equivalent of KafkaUtils.createRDD, i.e. read kafka topic to RDD by specifying offsets?
body: <p>How to read data in a kafka topic to an RDD by specifying start and end offsets?</p>
  
  <p><code>KafkaUtils.createRDD</code> <strike>is</strike> was experimental and the API is rather unpleasant (it returns a big bloated Java <code>ConsumerRecord</code> class that is not even serialisable and puts it in <code>KafkaRDD</code>, which overrides a lot of methods (like persist) to just throw an exception.</p>
  
  <p>What I would like, is a straightforward API like this:</p>
  
  <pre><code>case class Message(key: String, 
                     value: String, 
                     offset: Long, 
                     timestamp: Long)
  
  def readKafka(topic: String, offsetsByPartition: Map[Int, (Long, Long)])
               (config: KafkaConfig, sc: SparkContext): RDD[Message]
  </code></pre>
  
  <p>Or something similar where <code>key: Array[Byte]</code> and <code>value: Array[Byte]</code></p>