Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix gc max wait time #2790

Merged
merged 1 commit into from
Sep 27, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package com.pingcap.tispark.safepoint

import com.google.common.util.concurrent.ThreadFactoryBuilder
import com.pingcap.tikv.ClientSession
import org.apache.spark.sql.SparkSession
import org.slf4j.LoggerFactory
import org.tikv.common.exception.TiInternalException
import org.tikv.common.meta.TiTimestamp
Expand All @@ -29,7 +30,8 @@ case class ServiceSafePoint(
serviceId: String,
ttl: Long,
GCMaxWaitTime: Long,
clientSession: ClientSession) {
clientSession: ClientSession,
sparkSession: SparkSession) {

private val PD_UPDATE_SAFE_POINT_BACKOFF: Int = 20 * 1000
private final val logger = LoggerFactory.getLogger(getClass.getName)
Expand All @@ -38,6 +40,14 @@ case class ServiceSafePoint(
new ThreadFactoryBuilder().setNameFormat("serviceSafePoint-thread-%d").setDaemon(true).build)
service.scheduleAtFixedRate(
() => {
val now = clientSession.getTiKVSession.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)
}
if (minStartTs != Long.MaxValue) {
val safePoint = clientSession.getTiKVSession.getPDClient.updateServiceGCSafePoint(
serviceId,
Expand Down
3 changes: 2 additions & 1 deletion core/src/main/scala/org/apache/spark/sql/TiContext.scala
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,8 @@ class TiContext(val sparkSession: SparkSession) extends Serializable with Loggin
"tispark_" + UUID.randomUUID,
TiConfigConst.DEFAULT_GC_SAFE_POINT_TTL,
GCMaxWaitTime,
clientSession)
clientSession,
sparkSession)

sparkSession.sparkContext.addSparkListener(new SparkListener() {
override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,4 +17,4 @@ spark.tispark.jdbc.tls_enable=true
spark.tispark.jdbc.server_cert_store = /config/cert/jks/server-cert-store
spark.tispark.jdbc.server_cert_password = 12345678
spark.tispark.jdbc.client_cert_store = /config/cert/jks/client-keystore
spark.tispark.jdbc.client_cert_password = 123456
spark.tispark.jdbc.client_cert_password = 123456
Loading