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