title: Why does Spark Streaming fail with ClassCastException with repartitioned dstream when accessing offsets?
body: <p>In my Spark application I create a DStream from a Kafka topic in the following way:</p>
  
  <pre><code> KafkaUtils
    .createDirectStream[String, Array[Byte], StringDecoder, DefaultDecoder, (String, Array[Byte])](
      streamingContext,
      kafkaParams,
      offset.get,
      { message: MessageAndMetadata[String, Array[Byte]] =&gt; (message.key(), message.message()) }
    )
  </code></pre>
  
  <p>and later, i commit offset to Kafka topic using <em>asInstanceOf</em> function:</p>
  
  <pre><code>directKafkaStream.foreachRDD { rdd =&gt; 
   val offsetRanges = rdd.asInstanceOf[HasOffsetRanges]
   // offsetRanges.length = # of Kafka partitions being consumed
   ... }
  </code></pre>
  
  <p>In this case everything is ok, but if i repartion the dstream, when i try to commit offsets I have the following exception:</p>
  
  <blockquote>
    <p>java.lang.ClassCastException: org.apache.spark.rdd.MapPartitionsRDD cannot be cast to org.apache.spark.streaming.kafka.HasOffsetRanges repartition</p>
  </blockquote>
  
  <p>Can someone help me?</p>