Spark 2.3.0 выпуск каста класса потока kafka

Question

Я использую обновленную версию (предыдущая kafka 0.8.x, теперь 0.10.2.0) потоковой передачи kafka в приложении Stream Streaming. И я вижу эту странную проблему приведения классов:

Вот часть кода:

JavaDStream<ConsumerRecord<String, Event>> stream = (creation logic)
stream.foreachRDD(rdd -> {  // JavaRdd<ConsumerRecord<String, Event>>
rdd.map(consumerRecord -> parse(consumerRecord.value()) // parse the value into a pair of Event -> DataPoint
.map(dataPair -> processDataPair(dataPair))
.mapPartitions(iterator -> processPartitioned(iterator)));
//... (some other processing logic)
})

И на линии mapPartitions я получил эту ошибку:

java.lang.ClassCastException: *.*.$Event cannot be cast to org.apache.kafka.clients.consumer.ConsumerRecord
at org.apache.spark.api.java.JavaRDD$$anonfun$filter$1.apply(JavaRDD.scala:78)
at org.apache.spark.api.java.JavaRDD$$anonfun$filter$1.apply(JavaRDD.scala:78)
at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:463)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439)
at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:461)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:461)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$class.toString(Iterator.scala:1332)
at scala.collection.AbstractIterator.toString(Iterator.scala:1336)

Что приводит к разрыву потока. Похоже, mapPartitions не принимал тип вывода предыдущих преобразований. Не знаете, как я могу это исправить?

Заранее спасибо!

Редактировать: Судя по тому, что я могу сказать, более новая версия mapPartitions ВСЕГДА выполняется сначала, а не по конвейеру. после преобразования карты.