Github user xuanyuanking commented on a diff in the pull request:
https://github.com/apache/spark/pull/20689#discussion_r174667490
--- Diff:
external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaContinuousReader.scala
---
@@ -164,7 +164,15 @@ case class KafkaContinuousDataReaderFactory(
startOffset: Long,
kafkaParams: ju.Map[String, Object],
pollTimeoutMs: Long,
- failOnDataLoss: Boolean) extends DataReaderFactory[UnsafeRow] {
+ failOnDataLoss: Boolean) extends
ContinuousDataReaderFactory[UnsafeRow] {
+
+ override def createDataReaderWithOffset(offset: PartitionOffset):
DataReader[UnsafeRow] = {
+ val kafkaOffset = offset.asInstanceOf[KafkaSourcePartitionOffset]
+ assert(kafkaOffset.topicPartition == topicPartition)
--- End diff --
Got it.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
