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: [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]
