[
https://issues.apache.org/jira/browse/SPARK-25315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16634423#comment-16634423
]
Shixiong Zhu commented on SPARK-25315:
--------------------------------------
Kafka’s own configurations should be set with "kafka." prefix. "group.id" and
"auto.offset.reset" will be ignored.
In addition, after you add "kafka." prefix, you will see some error messages as
"group.id" or "auto.offset.reset" is not supported. They are documented here:
http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#kafka-specific-configurations
> setting "auto.offset.reset" to "earliest" has no effect in Structured
> Streaming with Spark 2.3.1 and Kafka 1.0
> --------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-25315
> URL: https://issues.apache.org/jira/browse/SPARK-25315
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 2.3.1
> Environment: Standalone; running in IDEA
> Reporter: Zhenhao Li
> Priority: Major
>
> The following code won't read from the beginning of the topic
> ```
> {code:java}
> val kafkaOptions = Map[String, String](
> "kafka.bootstrap.servers" -> KAFKA_BOOTSTRAP_SERVERS,
> "subscribe" -> TOPIC,
> "group.id" -> GROUP_ID,
> "auto.offset.reset" -> "earliest"
> )
> val myStream = sparkSession
> .readStream
> .format("kafka")
> .options(kafkaOptions)
> .load()
> .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
> myStream
> .writeStream
> .format("console")
> .start()
> .awaitTermination()
> {code}
> ```
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
