You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zifeif2 (via GitHub)" <gi...@apache.org> on 2026/10/08 01:08:10 UTC
[PR] [WIP][STREAMING] Reacquire Kafka consumer during offset recovery [spark]
zifeif2 opened a new pull request, #59285:
URL: https://github.com/apache/spark/pull/59285
### What changes were proposed in this pull request?
This PR reacquires a Kafka consumer after the out-of-range recovery path releases the old
consumer back to the pool. It also adds a regression test that closes the returned consumer
immediately to reproduce pool cleanup racing with recovery.
### Why are the changes needed?
When `KafkaDataConsumer` handles an `OffsetOutOfRangeException`, it releases its borrowed
consumer to discard cached state. With `failOnDataLoss=false`, recovery then used the released
local reference to query the available offset range and retry the fetch. The pool can close that
idle consumer or lend it to another task, so using the stale reference can fail with an
`IllegalStateException` or access a consumer no longer exclusively owned by the task.
Reacquiring through the pool restores exclusive ownership before recovery continues.
### Does this PR introduce _any_ user-facing change?
Yes. Kafka reads with `failOnDataLoss=false` no longer fail intermittently when pool cleanup
coincides with recovery from an out-of-range offset. There is no API change.
### How was this patch tested?
Added a regression test that forces pool cleanup when the invalid consumer is returned. The
targeted `KafkaDataConsumerSuite` test passed in an equivalent downstream Spark build:
```
KafkaDataConsumerSuite -- -z "reacquire consumer before recovering from an out-of-range offset"
```
`git diff --check` passed.
The Apache SBT invocation was also attempted, but this host could not resolve Maven Central while
loading the project, before compilation or test execution.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-6).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60076][SS] Reacquire Kafka consumer during offset recovery [spark]
Posted by "zifeif2 (via GitHub)" <gi...@apache.org>.
zifeif2 commented on code in PR #59285:
URL: https://github.com/apache/spark/pull/59285#discussion_r4250681610
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala:
##########
@@ -101,6 +101,39 @@ class KafkaDataConsumerSuite
assert(e.getCause === cause)
}
+ test("reacquire consumer before recovering from an out-of-range offset") {
+ val testTopic = "out-of-range-recovery-topic-" + Random.nextInt()
+ val testTopicPartition = new TopicPartition(testTopic, 0)
+ testUtils.createTopic(testTopic, 1)
+ testUtils.sendMessages(Seq(new RecordBuilder(testTopic, "value").build()))
+
+ // Closing a consumer as soon as it is returned deterministically reproduces an eviction
+ // racing with the out-of-range recovery path.
+ val closingConsumerPool = new InternalKafkaConsumerPool(new SparkConf()) {
+ override def returnObject(consumer: InternalKafkaConsumer): Unit = {
+ super.returnObject(consumer)
+ reset()
+ }
+ }
+ val consumer = new KafkaDataConsumer(
+ testTopicPartition,
+ getKafkaParams(),
+ closingConsumerPool,
+ fetchedDataPool)
+
+ try {
+ val originalConsumer = consumer.getOrRetrieveConsumer()
+ // The only record is at offset 0, so requesting offset 1 triggers out-of-range recovery.
+ val record = consumer.get(1, 2, 1000, failOnDataLoss = false)
Review Comment:
Fixed in b284900a391. The test now produces offsets 0 through 2, deletes records before offset 2, sets `auto.offset.reset=none`, calls `get(0, 3, ...)`, and asserts that recovery returns offset 2 with value `value-2` through a newly borrowed wrapper. I also ran the probe-only negative control: the offset-range probe succeeded, then the resumed fetch failed in `InternalKafkaConsumer.fetch` on the closed wrapper; the same case passes with the PR fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60076][SS] Reacquire Kafka consumer during offset recovery [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59285:
URL: https://github.com/apache/spark/pull/59285#discussion_r4235991793
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala:
##########
@@ -101,6 +101,39 @@ class KafkaDataConsumerSuite
assert(e.getCause === cause)
}
+ test("reacquire consumer before recovering from an out-of-range offset") {
+ val testTopic = "out-of-range-recovery-topic-" + Random.nextInt()
+ val testTopicPartition = new TopicPartition(testTopic, 0)
+ testUtils.createTopic(testTopic, 1)
+ testUtils.sendMessages(Seq(new RecordBuilder(testTopic, "value").build()))
+
+ // Closing a consumer as soon as it is returned deterministically reproduces an eviction
+ // racing with the out-of-range recovery path.
+ val closingConsumerPool = new InternalKafkaConsumerPool(new SparkConf()) {
+ override def returnObject(consumer: InternalKafkaConsumer): Unit = {
+ super.returnObject(consumer)
+ reset()
+ }
+ }
+ val consumer = new KafkaDataConsumer(
+ testTopicPartition,
+ getKafkaParams(),
+ closingConsumerPool,
+ fetchedDataPool)
+
+ try {
+ val originalConsumer = consumer.getOrRetrieveConsumer()
+ // The only record is at offset 0, so requesting offset 1 triggers out-of-range recovery.
+ val record = consumer.get(1, 2, 1000, failOnDataLoss = false)
Review Comment:
**Nit (P3):** This case only reaches the no-overlap branch. With the single record at offset 0, `get(1, 2, ...)` hits `OffsetOutOfRangeException`, `getEarliestAvailableOffsetBetween` sees `[0, 1)` and returns `UNKNOWN_OFFSET`, and the loop exits without another `fetchRecord`. So nothing checks that the fetches after the probe use the reacquired consumer. For example, `getEarliestAvailableOffsetBetween(getOrRetrieveConsumer(), ...)` with `consumer` left as a `val` would still pass both assertions. It would then hit the closed consumer as soon as recovery resumes at `range.earliest`, the usual retention case.
Could you add a case where recovery resumes inside `[offset, untilOffset)`? For example, produce a few records, delete the earliest ones through the admin client so the requested offset falls below the log start offset, then assert that `get()` returns the record at the new earliest offset. One catch: the suite's `getKafkaParams()` uses `auto.offset.reset=earliest`, which silently resets instead of throwing. The new case needs `none`, as `kafkaParamsForExecutors` sets it, so the out-of-range error reaches the handler.
**Verification:**
- **Regression:** With a probe-only variant of the fix (a reacquired consumer passed only to getEarliestAvailableOffsetBetween while later fetches use the released one), the resumable-recovery case fails with the closed-consumer error. With the PR's rebinding of the loop's consumer, it passes.
- **Behavior:** The resumable case actually enters the out-of-range handler, which releases the consumer to the closing pool, and then returns the record at the new earliest available offset rather than null. The existing no-overlap case still returns null with a reacquired consumer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60076][SS] Reacquire Kafka consumer during offset recovery [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db closed pull request #59285: [SPARK-60076][SS] Reacquire Kafka consumer during offset recovery
URL: https://github.com/apache/spark/pull/59285
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60076][SS] Reacquire Kafka consumer during offset recovery [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59285:
URL: https://github.com/apache/spark/pull/59285#discussion_r4253886836
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala:
##########
@@ -101,6 +102,55 @@ class KafkaDataConsumerSuite
assert(e.getCause === cause)
}
+ test("reacquire consumer before recovering from an out-of-range offset") {
+ val testTopic = "out-of-range-recovery-topic-" + Random.nextInt()
+ val testTopicPartition = new TopicPartition(testTopic, 0)
+ testUtils.createTopic(testTopic, 1)
+ testUtils.sendMessages((0 until 3).map { offset =>
+ new RecordBuilder(testTopic, s"value-$offset").build()
+ })
+
+ val admin = Admin.create(Map[String, Object](
+ AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> testUtils.brokerAddress).asJava)
+ try {
+ admin.deleteRecords(Map(
+ testTopicPartition -> RecordsToDelete.beforeOffset(2L)).asJava).all().get()
+ } finally {
+ admin.close()
+ }
+ assert(testUtils.getEarliestOffsets(Set(testTopic))(testTopicPartition) === 2L)
+
+ val kafkaParams = new ju.HashMap[String, Object](getKafkaParams())
+ kafkaParams.put(AUTO_OFFSET_RESET_CONFIG, "none")
+
+ // Closing a consumer as soon as it is returned deterministically reproduces an eviction
+ // racing with the out-of-range recovery path.
+ val closingConsumerPool = new InternalKafkaConsumerPool(new SparkConf()) {
+ override def returnObject(consumer: InternalKafkaConsumer): Unit = {
+ super.returnObject(consumer)
+ reset()
+ }
+ }
+ val consumer = new KafkaDataConsumer(
+ testTopicPartition,
+ kafkaParams,
+ closingConsumerPool,
+ fetchedDataPool)
+
+ try {
+ val originalConsumer = consumer.getOrRetrieveConsumer()
+ val record = consumer.get(0, 3, 1000, failOnDataLoss = false)
Review Comment:
**Nit (P3):** The new test passes a 1000 ms poll timeout here, while every other `get()` call in this suite uses 10000 ms. After recovery picks offset 2, `InternalKafkaConsumer.fetch` polls the newly created replacement consumer for only 1000 ms. If no record arrives in time on a slow runner, the position stays at 2, which is still inside `[2, 3)`, so `fetch` throws `TimeoutException`. `get()` only catches `OffsetOutOfRangeException`, so the test fails even though the fix is correct. Using the suite's usual 10000 ms costs nothing on the normal path, because `poll` returns as soon as the record is fetched.
```suggestion
val record = consumer.get(0, 3, 10000, failOnDataLoss = false)
```
**Verification:**
- **Regression:** The out-of-range recovery test still passes with the reacquire fix and still fails when recovery uses the released consumer for the probe or the resumed fetch.
- **Inspection:** The test's get() poll timeout matches the suite's 10000 ms convention, so a slow broker round trip on the resumed fetch cannot surface as a TimeoutException failure.
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala:
##########
@@ -101,6 +101,39 @@ class KafkaDataConsumerSuite
assert(e.getCause === cause)
}
+ test("reacquire consumer before recovering from an out-of-range offset") {
+ val testTopic = "out-of-range-recovery-topic-" + Random.nextInt()
+ val testTopicPartition = new TopicPartition(testTopic, 0)
+ testUtils.createTopic(testTopic, 1)
+ testUtils.sendMessages(Seq(new RecordBuilder(testTopic, "value").build()))
+
+ // Closing a consumer as soon as it is returned deterministically reproduces an eviction
+ // racing with the out-of-range recovery path.
+ val closingConsumerPool = new InternalKafkaConsumerPool(new SparkConf()) {
+ override def returnObject(consumer: InternalKafkaConsumer): Unit = {
+ super.returnObject(consumer)
+ reset()
+ }
+ }
+ val consumer = new KafkaDataConsumer(
+ testTopicPartition,
+ getKafkaParams(),
+ closingConsumerPool,
+ fetchedDataPool)
+
+ try {
+ val originalConsumer = consumer.getOrRetrieveConsumer()
+ // The only record is at offset 0, so requesting offset 1 triggers out-of-range recovery.
+ val record = consumer.get(1, 2, 1000, failOnDataLoss = false)
Review Comment:
Thanks, verified in b284900a391. With records before offset 2 deleted and `auto.offset.reset=none`, `get(0, 3, ...)` now goes through the out-of-range handler. The closing pool destroys the returned wrapper, and recovery resumes at offset 2 through the reborrowed consumer. A probe-only variant would now fail at the resumed fetch, so this covers the case I was after.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4235991793","thread_id":"inline:4235991793","verdict_sha256":"6b2dae72237d25fb154986ddc69be486d6a380eecdf26c757cdbb7d09f965125"} -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60076][STREAMING] Reacquire Kafka consumer during offset recovery [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #59285:
URL: https://github.com/apache/spark/pull/59285#discussion_r4235462415
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala:
##########
@@ -450,6 +450,9 @@ private[kafka010] class KafkaDataConsumer(
throwOnDataLoss(toFetchOffset, untilOffset, topicPartition, groupId, e)
} else {
logOnDataLoss(topicPartition, groupId, s"Cannot fetch offset $toFetchOffset", e)
+ // The consumer was released while handling the out-of-range offset, so reacquire it
Review Comment:
lets add the SPARK jira here and also explain that if we don't reacquire, the consumer could be released or reassigned leading to incorrect results or failures ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60076][SS] Reacquire Kafka consumer during offset recovery [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #59285:
URL: https://github.com/apache/spark/pull/59285#issuecomment-6175249807
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/84c89422b44c7a9f8cbd49a552d383c0c3be0438
- merged into branch-4.x https://github.com/apache/spark/commit/1bb793018d0299477f55cf063d01e9ab3306d500
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org