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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
