[SPARK-4964] [Streaming] Exactly-once semantics for Kafka

https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example

"Also note that we are explicitly checking that the offset being read is
not less than the offset that we requested. This is needed since if Kafka
is compressing the messages, the fetch request will return an entire
compressed block even if the requested offset isn't the beginning of the
compressed block. Thus a message we saw previously may be returned again."

On Tue, Dec 30, 2014 at 2:00 PM, Tathagata Das <notifications@github.com>
wrote:

> In external/kafka/src/main/scala/org/apache/spark/rdd/kafka/KafkaRDD.scala
> <https://github.com/apache/spark/pull/3798#discussion-diff-22362167>:
>
> > +              build()
> > +            val resp = consumer.fetch(req)
> > +            if (resp.hasError) {
> > +              val err = resp.errorCode(part.topic, part.partition)
> > +              if (err == ErrorMapping.LeaderNotAvailableCode ||
> > +                err == ErrorMapping.NotLeaderForPartitionCode) {
> > +                log.error(s"Lost leader for topic ${part.topic} partition ${part.partition}, " +
> > +                  s" sleeping for ${kc.config.refreshLeaderBackoffMs}ms")
> > +                Thread.sleep(kc.config.refreshLeaderBackoffMs)
> > +              }
> > +              // Let normal rdd retry sort out reconnect attempts
> > +              throw ErrorMapping.exceptionFor(err)
> > +            }
> > +            iter = resp.messageSet(part.topic, part.partition)
> > +              .iterator
> > +              .dropWhile(_.offset < requestOffset)
>
> Why is there a drop here? Doesnt the response return messages for the
> requested offset?
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/3798/files#r22362167>.
>