Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/4805#discussion_r26094221
--- Diff:
external/kafka/src/main/scala/org/apache/spark/streaming/kafka/DirectKafkaInputDStream.scala
---
@@ -118,6 +123,7 @@ class DirectKafkaInputDStream[
context.sparkContext, kafkaParams, currentOffsets, untilOffsets,
messageHandler)
currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)
+ offsetMap += ((validTime, currentOffsets))
--- End diff --
I don't think offsetMap is required at all. The offset ranges are stored in
the KafkaRDDs. Why not just pull up the offset from them, rather than
introducing another hashmap that needs to be managed/cleared?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
