[
https://issues.apache.org/jira/browse/SPARK-20050?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sasaki Toru updated SPARK-20050:
--------------------------------
Description:
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.
was:
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.
Issue Type: Bug (was: Improvement)
> 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: Bug
> 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: [email protected]
For additional commands, e-mail: [email protected]
