[jira] [Created] (SPARK-20050) Kafka 0.10 DirectStream doesn't commit last processed batch's offset when graceful shutdown

["Sasaki Toru (JIRA)" <ji...@apache.org>]

Sasaki Toru created SPARK-20050:
-----------------------------------

             Summary: Kafka 0.10 DirectStream doesn't commit last processed batch's offset when graceful shutdown
                 Key: SPARK-20050
                 URL: https://issues.apache.org/jira/browse/SPARK-20050
             Project: Spark
          Issue Type: Improvement
          Components: DStreams
    Affects Versions: 2.2.0
            Reporter: Sasaki Toru


I use Kafka 0.10 DirectStream with properties 'enable.auto.commit=false' and call 'DirectKafkaInputDStream#commitAsync' finally in each batches such below

{code}
val kafkaStream = KafkaUtils.createDirectStream[String, String](...)

kafkaStream.map { input =>
  "key: " + input.key.toString + " value: " + input.value.toString + " offset: " + input.offset.toString
  }.foreachRDD { rdd =>
    rdd.foreach { input =>
    println(input)
  }
}

kafkaStream.foreachRDD { rdd =>
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  kafkaStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
{\code}

Some records which processed in the last batch before Streaming graceful shutdown reprocess in the first batch after Spark Streaming restart.

It may cause offsets specified in commitAsync will commit in the head of next batch.




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org