cp (#2792)

pingcap · Oct 8, 2024 · 8f7901e · 8f7901e
1 parent a437268
commit 8f7901e
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 2 deletions.
diff --git a/core/src/main/scala/com/pingcap/tispark/safepoint/ServiceSafePoint.scala b/core/src/main/scala/com/pingcap/tispark/safepoint/ServiceSafePoint.scala
@@ -23,6 +23,7 @@ import com.pingcap.tikv.exception.TiInternalException
 import com.pingcap.tikv.meta.TiTimestamp
 import com.pingcap.tikv.util.{BackOffer, ConcreteBackOffer}
 import com.google.common.util.concurrent.ThreadFactoryBuilder
+import org.apache.spark.sql.SparkSession
 import org.slf4j.LoggerFactory
 
 import java.util.concurrent.{Executors, ScheduledExecutorService, TimeUnit}
@@ -31,7 +32,8 @@ case class ServiceSafePoint(
     serviceId: String,
     ttl: Long,
     GCMaxWaitTime: Long,
-    tiSession: TiSession) {
+    tiSession: TiSession,
+    sparkSession: SparkSession) {
 
   private final val logger = LoggerFactory.getLogger(getClass.getName)
   private var minStartTs = Long.MaxValue
@@ -40,6 +42,14 @@ case class ServiceSafePoint(
   service.scheduleAtFixedRate(
     () => {
       if (minStartTs != Long.MaxValue) {
+        val now = tiSession.getTimestamp
+        if (now.getPhysical - TiTimestamp.extractPhysical(minStartTs) >= GCMaxWaitTime * 1000) {
+          val msg =
+            s"Can not pause GC more than spark.tispark.gc_max_wait_time=$GCMaxWaitTime s. start_ts: ${minStartTs}, now: ${now.getVersion}. You can adjust spark.tispark.gc_max_wait_time to increase the gc max wait time."
+          logger.error(msg)
+          sparkSession.stop()
+          throw new TiInternalException(msg)
+        }
         val safePoint = tiSession.getPDClient.updateServiceGCSafePoint(
           serviceId,
           ttl,

diff --git a/core/src/main/scala/org/apache/spark/sql/TiContext.scala b/core/src/main/scala/org/apache/spark/sql/TiContext.scala
@@ -65,7 +65,8 @@ class TiContext(val sparkSession: SparkSession) extends Serializable with Loggin
       "tispark_" + UUID.randomUUID,
       TiConfigConst.DEFAULT_GC_SAFE_POINT_TTL,
       GCMaxWaitTime,
-      tiSession)
+      tiSession,
+      sparkSession)
 
   sparkSession.sparkContext.addSparkListener(new SparkListener() {
     override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {