[
https://issues.apache.org/jira/browse/SPARK-49123?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Cheng Pan resolved SPARK-49123.
-------------------------------
Fix Version/s: 4.3.0
Resolution: Fixed
Issue resolved by pull request 57297
[https://github.com/apache/spark/pull/57297]
> Improve the logging behavior when Spark loaded zero result from Kafka after
> poll timeout
> ----------------------------------------------------------------------------------------
>
> Key: SPARK-49123
> URL: https://issues.apache.org/jira/browse/SPARK-49123
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 3.3.0
> Environment: java version: openjdk version "1.8.0_402"
> python version: 3.10.2
> os: amazon linux 2
>
>
> Reporter: Yang Guo
> Priority: Major
> Labels: kafka, pull-request-available, sql
> Fix For: 4.3.0
>
> Original Estimate: 48h
> Remaining Estimate: 48h
>
> When using spark structured streaming to read data from Kafka, the poll
> request can time out if the producer is writing data to the topic in
> transaction mode but not commit the transaction for a long time.
>
> When this happened, the `r.isEmpty` as `True` but the offsetAfterPoll
> changed. This will not thrown any error or warning logs. This makes users
> have no idea why the related spark task delay for `pollTimeoutMs`. Here is
> the related
> [code|https://github.com/apache/spark/blob/v3.3.0/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala#L85-L101]:
>
>
> {code:java}
> if (r.isEmpty) {
> // We cannot fetch anything after `poll`. Two possible cases:
> // - `offset` is out of range so that Kafka returns nothing.
> `OffsetOutOfRangeException` will
> // be thrown.
> // - Cannot fetch any data before timeout. `TimeoutException` will be
> thrown.
> // - Fetched something but all of them are not invisible. This is a
> valid case and let the
> // caller handles this.
> if (offset < range.earliest || offset >= range.latest) {
> throw new OffsetOutOfRangeException(
> Map(topicPartition -> java.lang.Long.valueOf(offset)).asJava)
> } else if (offset == offsetAfterPoll) {
> throw new TimeoutException(
> s"Cannot fetch record for offset $offset in $pollTimeoutMs
> milliseconds")
> }
> }{code}
>
>
> To improve the user experience for troubleshooting when the poll get zero
> result before timeout but offsetAfterPoll changed, there should add some
> warning or error level logs:
>
> A sample looks like the following:
>
> {code:java}
> if (r.isEmpty) {
> // We cannot fetch anything after `poll`. Two possible cases:
> // - `offset` is out of range so that Kafka returns nothing.
> `OffsetOutOfRangeException` will
> // be thrown.
> // - Cannot fetch any data before timeout. `TimeoutException` will be
> thrown.
> // - Fetched something but all of them are invisible. This is a valid
> case and let the
> // caller handles this. Although this is a valid case, it is
> necessary to add warning level logs to indicate zero data returned before
> timeout.
> if (offset < range.earliest || offset >= range.latest) {
> throw new OffsetOutOfRangeException(
> Map(topicPartition -> java.lang.Long.valueOf(offset)).asJava)
> } else if (offset == offsetAfterPoll) {
> throw new TimeoutException(
> s"Cannot fetch record for offset $offset in $pollTimeoutMs
> milliseconds")
> } else {
> logWarning(s"After taking $pollTimeoutMs milliseconds polling, ZERO
> record fetched while offset changed from $offset to $offsetAfterPoll.")
> }
> } {code}
>
>
> By the way, fix the typo from `Fetched something but all of them are not
> invisible` to `Fetched something but all of them are invisible`.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
