vinodkc commented on code in PR #57351:
URL: https://github.com/apache/spark/pull/57351#discussion_r3615721162
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderSuite.scala:
##########
@@ -323,4 +323,90 @@ class KafkaOffsetReaderSuite extends SharedSparkSession
with KafkaTest {
reader.close()
}
}
+
+ // SPARK-49442: When partition.metadata.cache.ttl.ms is set, repeated
fetchLatestOffsets
+ // calls within the TTL window must reuse the cached partition set and issue
only one
+ // DescribeTopics RPC to the broker, regardless of how many micro-batches
run.
+ test("SPARK-49442: partition.metadata.cache.ttl.ms suppresses redundant " +
+ "DescribeTopics RPCs within the TTL window") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 3)
+
+ val ttlMs = 300000 // 5 minutes
+ val describeCount = new AtomicInteger(0)
+ val countingStrategy = new SubscribeStrategy(Seq(topic)) {
+ override def assignedTopicPartitions(admin: Admin): Set[TopicPartition]
= {
+ describeCount.incrementAndGet()
+ super.assignedTopicPartitions(admin)
+ }
+ }
+
+ val reader = new KafkaOffsetReaderAdmin(
+ countingStrategy,
+ KafkaSourceProvider.kafkaParamsForDriver(Map(
+ "bootstrap.servers" -> testUtils.brokerAddress
+ )),
+ CaseInsensitiveMap(Map(
+ KafkaSourceProvider.PARTITION_METADATA_CACHE_TTL_MS -> ttlMs.toString
+ )),
+ ""
+ )
+
+ try {
+ val numBatches = 5
+ for (_ <- 1 to numBatches) {
+ reader.fetchLatestOffsets(None)
+ }
+ // All 5 fetches fall within the TTL window, so the partition set is
resolved only once.
+ assert(describeCount.get() === 1,
+ s"Expected 1 DescribeTopics call but got ${describeCount.get()}")
+ } finally {
+ reader.close()
+ }
+ }
+
+ test("SPARK-49442: partition.metadata.cache.ttl.ms refreshes after TTL
expires") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 3)
+
+ val ttlMs = 200 // short TTL so the test doesn't need to sleep long
Review Comment:
Thanks for the review comments.
Changed to 2000ms.
--
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]
