[SPARK-34187][SS] Use available offset range obtained during polling when checking offset validation #31275
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This comment has been minimized.
This comment has been minimized.
...kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala
Outdated
Show resolved
Hide resolved
|
Kubernetes integration test starting |
|
Test build #134340 has finished for PR 31275 at commit
|
|
Kubernetes integration test status failure |
| @@ -192,6 +197,13 @@ private[consumer] case class FetchedData( | |||
| * Returns the next offset to poll after draining the pre-fetched records. | |||
| */ | |||
| def offsetAfterPoll: Long = _offsetAfterPoll | |||
|
|
|||
| /** | |||
| * Returns the tuple of earliest and latest offsets that is the available offset range when | |||
There was a problem hiding this comment.
nit: Use @returns annotation.
Sorry for being late with this, but I've just noticed it.
There was a problem hiding this comment.
Hm, I think this just follows other methods above and below.
There was a problem hiding this comment.
Given your assessment, could you make it as a correctness issue, @viirya ?
It looks not correct ...
... drop the record.
...l/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/FetchedDataPool.scala
Outdated
Show resolved
Hide resolved
dongjoon-hyun
approved these changes
Jan 24, 2021
Done. Thanks. |
|
Thank you for update, @viirya . |
dongjoon-hyun
pushed a commit
that referenced
this pull request
Jan 24, 2021
…when checking offset validation ### What changes were proposed in this pull request? This patch uses the available offset range obtained during polling Kafka records to do offset validation check. ### Why are the changes needed? We support non-consecutive offsets for Kafka since 2.4.0. In `fetchRecord`, we do offset validation by checking if the offset is in available offset range. But currently we obtain latest available offset range to do the check. It looks not correct as the available offset range could be changed during the batch, so the available offset range is different than the one when we polling the records from Kafka. It is possible that an offset is valid when polling, but at the time we do the above check, it is out of latest available offset range. We will wrongly consider it as data loss case and fail the query or drop the record. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This should pass existing unit tests. This is hard to have unit test as the Kafka producer and the consumer is asynchronous. Further, we also need to make the offset out of new available offset range. Closes #31275 from viirya/SPARK-34187. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com> (cherry picked from commit ab6c0e5) Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
|
Merged to master/3.1. Could you make a backport to 3.0/2.4? |
|
Test build #134414 has finished for PR 31275 at commit
|
|
Thanks @dongjoon-hyun. Will file backport later. |
|
Nice finding, and thanks for the fix! |
skestle
pushed a commit
to skestle/spark
that referenced
this pull request
Feb 3, 2021
…when checking offset validation ### What changes were proposed in this pull request? This patch uses the available offset range obtained during polling Kafka records to do offset validation check. ### Why are the changes needed? We support non-consecutive offsets for Kafka since 2.4.0. In `fetchRecord`, we do offset validation by checking if the offset is in available offset range. But currently we obtain latest available offset range to do the check. It looks not correct as the available offset range could be changed during the batch, so the available offset range is different than the one when we polling the records from Kafka. It is possible that an offset is valid when polling, but at the time we do the above check, it is out of latest available offset range. We will wrongly consider it as data loss case and fail the query or drop the record. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This should pass existing unit tests. This is hard to have unit test as the Kafka producer and the consumer is asynchronous. Further, we also need to make the offset out of new available offset range. Closes apache#31275 from viirya/SPARK-34187. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This patch uses the available offset range obtained during polling Kafka records to do offset validation check.
Why are the changes needed?
We support non-consecutive offsets for Kafka since 2.4.0. In
fetchRecord, we do offset validation by checking if the offset is in available offset range. But currently we obtain latest available offset range to do the check. It looks not correct as the available offset range could be changed during the batch, so the available offset range is different than the one when we polling the records from Kafka.It is possible that an offset is valid when polling, but at the time we do the above check, it is out of latest available offset range. We will wrongly consider it as data loss case and fail the query or drop the record.
Does this PR introduce any user-facing change?
No
How was this patch tested?
This should pass existing unit tests.
This is hard to have unit test as the Kafka producer and the consumer is asynchronous. Further, we also need to make the offset out of new available offset range.