Skip to content
New issue

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

#

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

Already on GitHub? # to your account

Jump to bottom

[SPARK-49824][SS][CONNECT] Improve logging in SparkConnectStreamingQueryCache #48293

Closed
wants to merge 3 commits into from
Closed

[SPARK-49824][SS][CONNECT] Improve logging in SparkConnectStreamingQueryCache #48293

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
100a177
done, pending CI
WweiL Sep 28, 2024
35db862
retrigger
WweiL Sep 30, 2024
8120e80
retrigger
WweiL Sep 30, 2024
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
18 changes: 12 additions & 6 deletions ...src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamingQueryCache.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import scala.concurrent.duration.{Duration, DurationInt, FiniteDuration}
import scala.util.control.NonFatal

import org.apache.spark.internal.{Logging, MDC}
import org.apache.spark.internal.LogKeys.{DURATION, NEW_VALUE, OLD_VALUE, QUERY_CACHE_VALUE, QUERY_ID, SESSION_ID}
import org.apache.spark.internal.LogKeys.{DURATION, NEW_VALUE, OLD_VALUE, QUERY_CACHE_VALUE, QUERY_ID, QUERY_RUN_ID, SESSION_ID}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.StreamingQuery
import org.apache.spark.util.{Clock, SystemClock, ThreadUtils}
Expand Down Expand Up @@ -158,7 +158,8 @@ private[connect] class SparkConnectStreamingQueryCache(
if (v.userId.equals(sessionHolder.userId) && v.sessionId.equals(sessionHolder.sessionId)) {
if (v.query.isActive && Option(v.session.streams.get(k.queryId)).nonEmpty) {
logInfo(
log"Stopping the query with id ${MDC(QUERY_ID, k.queryId)} " +
log"Stopping the query with id: ${MDC(QUERY_ID, k.queryId)} " +
log"runId: ${MDC(QUERY_RUN_ID, k.runId)} " +
log"since the session has timed out")
try {
if (blocking) {
Expand All @@ -170,7 +171,8 @@ private[connect] class SparkConnectStreamingQueryCache(
} catch {
case NonFatal(ex) =>
logWarning(
log"Failed to stop the query ${MDC(QUERY_ID, k.queryId)}. " +
log"Failed to stop the with id: ${MDC(QUERY_ID, k.queryId)} " +
log"runId: ${MDC(QUERY_RUN_ID, k.runId)} " +
log"Error is ignored.",
ex)
}
Expand Down Expand Up @@ -238,25 +240,29 @@ private[connect] class SparkConnectStreamingQueryCache(

for ((k, v) <- queryCache) {
val id = k.queryId
val runId = k.runId
v.expiresAtMs match {

case Some(ts) if nowMs >= ts => // Expired. Drop references.
logInfo(
log"Removing references for ${MDC(QUERY_ID, id)} in " +
log"Removing references for id: ${MDC(QUERY_ID, id)} " +
log"runId: ${MDC(QUERY_RUN_ID, runId)} in " +
log"session ${MDC(SESSION_ID, v.sessionId)} after expiry period")
queryCache.remove(k)

case Some(_) => // Inactive query waiting for expiration. Do nothing.
logInfo(
log"Waiting for the expiration for ${MDC(QUERY_ID, id)} in " +
log"Waiting for the expiration for id: ${MDC(QUERY_ID, id)} " +
log"runId: ${MDC(QUERY_RUN_ID, runId)} in " +
log"session ${MDC(SESSION_ID, v.sessionId)}")

case None => // Active query, check if it is stopped. Enable timeout if it is stopped.
val isActive = v.query.isActive && Option(v.session.streams.get(id)).nonEmpty

if (!isActive) {
logInfo(
log"Marking query ${MDC(QUERY_ID, id)} in " +
log"Marking query id: ${MDC(QUERY_ID, id)} " +
log"runId: ${MDC(QUERY_RUN_ID, runId)} in " +
log"session ${MDC(SESSION_ID, v.sessionId)} inactive.")
val expiresAtMs = nowMs + stoppedQueryInactivityTimeout.toMillis
queryCache.put(k, v.copy(expiresAtMs = Some(expiresAtMs)))
Expand Down