(spark) branch branch-4.x updated: [SPARK-49442][SS][FOLLOWUP] Use monotonic clock for Kafka partition cache TTL

[ka...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 130cb0714b07 [SPARK-49442][SS][FOLLOWUP] Use monotonic clock for Kafka partition cache TTL
130cb0714b07 is described below

commit 130cb0714b0794112292cab310ae07b9ad4c60e5
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Thu Aug 20 12:50:09 2026 +0900

    [SPARK-49442][SS][FOLLOWUP] Use monotonic clock for Kafka partition cache TTL
    
    ### What changes were proposed in this pull request?
    
    Followup to https://github.com/apache/spark/pull/57351.
    
    This PR measures the Kafka partition metadata cache TTL with `System.nanoTime` instead of
    `System.currentTimeMillis`. It also represents the cached timestamp with `Option[Long]` rather than
    using zero as a sentinel. Debug logging continues to report cache age in milliseconds.
    
    ### Why are the changes needed?
    
    TTL measurement should use a monotonic time source. Wall-clock adjustments can otherwise make a
    cache entry appear older or younger than it is, causing premature refreshes or extending its
    lifetime. `System.nanoTime` measures elapsed time without that sensitivity.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No. The cache option and its behavior under a stable clock are unchanged.
    
    ### How was this patch tested?
    
    Existing cache reuse and expiry coverage was run with Java 17:
    
    ```
    build/sbt 'sql-kafka-0-10/testOnly org.apache.spark.sql.kafka010.KafkaOffsetReaderSuite'
    ```
    
    All 22 tests passed. No new test was added because the change only substitutes the JVM's monotonic
    elapsed-time source; reliably simulating host wall-clock changes is not practical in this suite.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Codex (GPT-5)
    
    Closes #57943 from cloud-fan/SPARK-49442-followup.
    
    Authored-by: Wenchen Fan <we...@databricks.com>
    Signed-off-by: Jungtaek Lim <ka...@gmail.com>
    (cherry picked from commit a37e9eafc242aa7988ceebae94e09d209c610bb7)
    Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
 .../sql/kafka010/KafkaOffsetReaderAdmin.scala      | 29 +++++++++++++---------
 1 file changed, 17 insertions(+), 12 deletions(-)

diff --git a/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala b/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala
index f182e156c76e..dc22a46e17bc 100644
--- a/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala
+++ b/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.kafka010
 
 import java.{util => ju}
 import java.util.Locale
+import java.util.concurrent.TimeUnit
 
 import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters._
@@ -62,7 +63,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
 
   // Protected by this.synchronized (always accessed inside withRetries, which holds the lock).
   private var cachedPartitions: Set[TopicPartition] = Set.empty
-  private var cacheTimestampMs: Long = 0L
+  private var cacheTimestampNanos: Option[Long] = None
 
   /**
    * An AdminClient used in the driver to query the latest Kafka offsets.
@@ -451,16 +452,20 @@ private[kafka010] class KafkaOffsetReaderAdmin(
     if (partitionMetadataCacheTtlMs <= 0) {
       return consumerStrategy.assignedTopicPartitions(admin)
     }
-    val now = System.currentTimeMillis()
-    if (cacheTimestampMs > 0 && (now - cacheTimestampMs) < partitionMetadataCacheTtlMs) {
-      logDebug(s"Reusing cached partitions (age ${now - cacheTimestampMs}ms < " +
-        s"${partitionMetadataCacheTtlMs}ms TTL): $cachedPartitions")
-      cachedPartitions
-    } else {
-      val fresh = consumerStrategy.assignedTopicPartitions(admin)
-      cachedPartitions = fresh
-      cacheTimestampMs = now
-      fresh
+    val nowNanos = System.nanoTime()
+    cacheTimestampNanos match {
+      case Some(timestampNanos)
+          if nowNanos - timestampNanos <
+            TimeUnit.MILLISECONDS.toNanos(partitionMetadataCacheTtlMs) =>
+        val cacheAgeMs = TimeUnit.NANOSECONDS.toMillis(nowNanos - timestampNanos)
+        logDebug(s"Reusing cached partitions (age ${cacheAgeMs}ms < " +
+          s"${partitionMetadataCacheTtlMs}ms TTL): $cachedPartitions")
+        cachedPartitions
+      case _ =>
+        val fresh = consumerStrategy.assignedTopicPartitions(admin)
+        cachedPartitions = fresh
+        cacheTimestampNanos = Some(nowNanos)
+        fresh
     }
   }
 
@@ -519,6 +524,6 @@ private[kafka010] class KafkaOffsetReaderAdmin(
     stopAdmin()
     _admin = null  // will automatically get reinitialized again
     cachedPartitions = Set.empty
-    cacheTimestampMs = 0L
+    cacheTimestampNanos = None
   }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org