You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2023/07/17 04:36:16 UTC
[spark] branch master updated: [SPARK-44439][CONNECT][SS] Fixed listListeners to only send ids back to client
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 61b9236654d [SPARK-44439][CONNECT][SS] Fixed listListeners to only send ids back to client
61b9236654d is described below
commit 61b9236654d863bf675d85c3000924d7499f8a4a
Author: bogao007 <bo...@databricks.com>
AuthorDate: Mon Jul 17 13:36:00 2023 +0900
[SPARK-44439][CONNECT][SS] Fixed listListeners to only send ids back to client
### What changes were proposed in this pull request?
This PR fixes listListener that's being implemented in a previous [PR](https://github.com/apache/spark/pull/41752) that sends listener class payload to client. We will only send ids as references of listener classes to client side instead.
### Why are the changes needed?
We should only return what users added previously in `listListeners` API. Only sending ids would be much more light-weighted.
### Does this PR introduce _any_ user-facing change?
yes
### How was this patch tested?
Unit tested.
Also performed manual integration test
```
import scala.collection.mutable
import scala.collection.mutable
import org.apache.spark.sql.streaming.StreamingQueryProgress
import org.apache.spark.sql.streaming.StreamingQueryProgress
import org.apache.spark.sql.streaming.StreamingQueryListener
import org.apache.spark.sql.streaming.StreamingQueryListener
import org.apache.spark.sql.streaming.StreamingQueryListener.{QueryIdleEvent, QueryStartedEvent, QueryTerminatedEvent}
import org.apache.spark.sql.streaming.StreamingQueryListener.{QueryIdleEvent, QueryStartedEvent, QueryTerminatedEvent}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession
class EventCollector extends StreamingQueryListener {
volatile var startEvent: QueryStartedEvent = null
volatile var terminationEvent: QueryTerminatedEvent = null
volatile var idleEvent: QueryIdleEvent = null
private val _progressEvents = new mutable.Queue[StreamingQueryProgress]
def progressEvents: Seq[StreamingQueryProgress] = _progressEvents.synchronized {
_progressEvents.clone().toSeq
}
override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
startEvent = event
val spark = SparkSession.builder().getOrCreate()
val df = spark.createDataFrame(Seq((1, 2), (4, 5)))
df.write.saveAsTable("my_listener_table")
}
override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
_progressEvents += event.progress
}
override def onQueryIdle(event: StreamingQueryListener.QueryIdleEvent): Unit = {
idleEvent = event
}
override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
terminationEvent = event
}
}
defined class EventCollector
spark.streams.listListeners()
res6: Array[StreamingQueryListener] = Array()
val listener = new EventCollector
listener: EventCollector = ammonite.$sess.cmd5$Helper$EventCollector1cd4082a
spark.streams.addListener(listener)
spark.streams.listListeners()
res9: Array[StreamingQueryListener] = Array(ammonite.$sess.cmd5$Helper$EventCollector1cd4082a)
spark.streams.listListeners()
res10: Array[StreamingQueryListener] = Array(ammonite.$sess.cmd5$Helper$EventCollector1cd4082a)
val listener1 = new EventCollector
listener1: EventCollector = ammonite.$sess.cmd5$Helper$EventCollector3c251258
spark.streams.addListener(listener1)
spark.streams.listListeners()
res13: Array[StreamingQueryListener] = Array(ammonite.$sess.cmd5$Helper$EventCollector1cd4082a, ammonite.$sess.cmd5$Helper$EventCollector3c251258)
spark.streams.listListeners()
res14: Array[StreamingQueryListener] = Array(ammonite.$sess.cmd5$Helper$EventCollector1cd4082a, ammonite.$sess.cmd5$Helper$EventCollector3c251258)
val q = spark.readStream.format("rate").load().writeStream.format("console").start()
q: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.streaming.RemoteStreamingQuery42be0eca
q.stop()
spark.streams.listListeners()
res17: Array[StreamingQueryListener] = Array(ammonite.$sess.cmd5$Helper$EventCollector1cd4082a, ammonite.$sess.cmd5$Helper$EventCollector3c251258)
spark.table("my_listener_table").show()
23/07/14 16:41:36 INFO CodeGenerator: Code generated in 98.321167 ms
+---+---+
| _1| _2|
+---+---+
| 4| 5|
| 1| 2|
+---+---+
```
Closes #42013 from bogao007/list-listener.
Authored-by: bogao007 <bo...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../sql/streaming/StreamingQueryManager.scala | 13 +-
.../ui/StreamingQueryStatusListener.scala | 166 ---------------------
.../apache/spark/sql/streaming/ui/UIUtils.scala | 76 ----------
.../CheckConnectJvmClientCompatibility.scala | 19 +--
.../spark/sql/streaming/StreamingQuerySuite.scala | 20 +--
.../src/main/protobuf/spark/connect/commands.proto | 3 +-
.../sql/connect/planner/SparkConnectPlanner.scala | 19 +--
.../spark/sql/connect/service/SessionHolder.scala | 7 +
python/pyspark/sql/connect/proto/commands_pb2.py | 20 +--
python/pyspark/sql/connect/proto/commands_pb2.pyi | 16 +-
.../ui/StreamingQueryStatusListener.scala | 4 +-
11 files changed, 42 insertions(+), 321 deletions(-)
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala
index f22adc1a966..8f9e768d23f 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala
@@ -179,15 +179,10 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession) extends Lo
* @since 3.5.0
*/
def listListeners(): Array[StreamingQueryListener] = {
- executeManagerCmd(_.setListListeners(true)).getListListeners.getListenersList.asScala.map {
- listener =>
- Utils
- .deserialize[StreamingListenerPacket](
- listener.getListenerPayload.toByteArray,
- Utils.getContextOrSparkClassLoader)
- .listener
- .asInstanceOf[StreamingQueryListener]
- }.toArray
+ executeManagerCmd(_.setListListeners(true)).getListListeners.getListenerIdsList.asScala
+ .filter(listenerCache.containsKey(_))
+ .map(listenerCache.get(_))
+ .toArray
}
private def executeManagerCmd(
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
deleted file mode 100644
index 113522ea536..00000000000
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
+++ /dev/null
@@ -1,166 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.streaming.ui
-
-import java.util.UUID
-import java.util.concurrent.ConcurrentHashMap
-
-import scala.collection.mutable
-
-import com.fasterxml.jackson.annotation.JsonIgnore
-
-import org.apache.spark.SparkConf
-import org.apache.spark.sql.internal.StaticSQLConf
-import org.apache.spark.sql.streaming.{StreamingQueryListener, StreamingQueryProgress}
-import org.apache.spark.sql.streaming.ui.StreamingQueryProgressWrapper._
-import org.apache.spark.sql.streaming.ui.UIUtils.parseProgressTimestamp
-import org.apache.spark.status.{ElementTrackingStore, KVUtils}
-import org.apache.spark.status.KVUtils.KVIndexParam
-import org.apache.spark.util.kvstore.KVIndex
-
-/**
- * A customized StreamingQueryListener used in structured streaming UI, which contains all UI data
- * for both active and inactive query.
- */
-private[sql] class StreamingQueryStatusListener(
- @transient val conf: SparkConf,
- @transient val store: ElementTrackingStore)
- extends StreamingQueryListener {
-
- private val streamingProgressRetention =
- conf.get(StaticSQLConf.STREAMING_UI_RETAINED_PROGRESS_UPDATES)
- private val inactiveQueryStatusRetention = conf.get(StaticSQLConf.STREAMING_UI_RETAINED_QUERIES)
-
- store.addTrigger(classOf[StreamingQueryData], inactiveQueryStatusRetention) { count =>
- cleanupInactiveQueries(count)
- }
-
- // Events from the same query run will never be processed concurrently, so it's safe to
- // access `progressIds` without any protection.
- private val queryToProgress = new ConcurrentHashMap[UUID, mutable.Queue[String]]()
-
- private def cleanupInactiveQueries(count: Long): Unit = {
- val view = store.view(classOf[StreamingQueryData]).index("active").first(false).last(false)
- val inactiveQueries = KVUtils.viewToSeq(view)
- val numInactiveQueries = inactiveQueries.size
- if (numInactiveQueries <= inactiveQueryStatusRetention) {
- return
- }
- val toDelete = inactiveQueries
- .sortBy(_.endTimestamp.get)
- .take(numInactiveQueries - inactiveQueryStatusRetention)
- val runIds = toDelete.map { e =>
- store.delete(e.getClass, e.runId)
- e.runId
- }
- // Delete wrappers in one pass, as deleting them for each summary is slow
- store.removeAllByIndexValues(classOf[StreamingQueryProgressWrapper], "runId", runIds)
- }
-
- override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
- val startTimestamp = parseProgressTimestamp(event.timestamp)
- store.write(
- new StreamingQueryData(
- event.name,
- event.id,
- event.runId.toString,
- isActive = true,
- None,
- startTimestamp),
- checkTriggers = true)
- }
-
- override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
- val runId = event.progress.runId
- val batchId = event.progress.batchId
- val timestamp = event.progress.timestamp
- if (!queryToProgress.containsKey(runId)) {
- queryToProgress.put(runId, mutable.Queue.empty[String])
- }
- val progressIds = queryToProgress.get(runId)
- progressIds.enqueue(getUniqueId(runId, batchId, timestamp))
- store.write(new StreamingQueryProgressWrapper(event.progress))
- while (progressIds.length > streamingProgressRetention) {
- val uniqueId = progressIds.dequeue
- store.delete(classOf[StreamingQueryProgressWrapper], uniqueId)
- }
- }
-
- override def onQueryIdle(event: StreamingQueryListener.QueryIdleEvent): Unit = {}
-
- override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
- val querySummary = store.read(classOf[StreamingQueryData], event.runId.toString)
- val curTime = System.currentTimeMillis()
- store.write(
- new StreamingQueryData(
- querySummary.name,
- querySummary.id,
- querySummary.runId,
- isActive = false,
- event.exception,
- querySummary.startTimestamp,
- Some(curTime)),
- checkTriggers = true)
- queryToProgress.remove(event.runId)
- }
-}
-
-private[spark] class StreamingQueryData(
- val name: String,
- val id: UUID,
- @KVIndexParam val runId: String,
- @KVIndexParam("active") val isActive: Boolean,
- val exception: Option[String],
- @KVIndexParam("startTimestamp") val startTimestamp: Long,
- val endTimestamp: Option[Long] = None)
-
-/**
- * This class contains all message related to UI display, each instance corresponds to a single
- * [[org.apache.spark.sql.streaming.StreamingQuery]].
- */
-private[sql] case class StreamingQueryUIData(
- summary: StreamingQueryData,
- recentProgress: Array[StreamingQueryProgress]) {
-
- def lastProgress: StreamingQueryProgress = {
- if (recentProgress.nonEmpty) {
- recentProgress.last
- } else {
- null
- }
- }
-}
-
-private[spark] class StreamingQueryProgressWrapper(val progress: StreamingQueryProgress) {
- @JsonIgnore @KVIndex
- private val uniqueId: String = getUniqueId(progress.runId, progress.batchId, progress.timestamp)
-
- @JsonIgnore @KVIndex("runId")
- private def runIdIndex: String = progress.runId.toString
-}
-
-private[sql] object StreamingQueryProgressWrapper {
-
- /**
- * Adding `timestamp` into unique id to support reporting `empty` query progress in which no
- * data comes but with the same batchId.
- */
- def getUniqueId(runId: UUID, batchId: Long, timestamp: String): String = {
- s"${runId}_${batchId}_$timestamp"
- }
-}
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/ui/UIUtils.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/ui/UIUtils.scala
deleted file mode 100644
index 88a110fa9a3..00000000000
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/ui/UIUtils.scala
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.streaming.ui
-
-import java.text.SimpleDateFormat
-
-import org.apache.spark.sql.catalyst.util.DateTimeUtils.getTimeZone
-
-private[ui] object UIUtils {
-
- /**
- * Check whether `number` is valid, if not return 0.0d
- */
- def withNumberInvalid(number: => Double): Double = {
- if (number.isNaN || number.isInfinite) {
- 0.0d
- } else {
- number
- }
- }
-
- /**
- * Execute a block of code when there is already one completed batch in streaming query,
- * otherwise return `default` value.
- */
- def withNoProgress[T](query: StreamingQueryUIData, body: => T, default: T): T = {
- if (query.lastProgress != null) {
- body
- } else {
- default
- }
- }
-
- def getQueryName(uiData: StreamingQueryUIData): String = {
- if (uiData.summary.name == null || uiData.summary.name.isEmpty) {
- "<no name>"
- } else {
- uiData.summary.name
- }
- }
-
- def getQueryStatus(uiData: StreamingQueryUIData): String = {
- if (uiData.summary.isActive) {
- "RUNNING"
- } else {
- uiData.summary.exception.map(_ => "FAILED").getOrElse("FINISHED")
- }
- }
-
- private val progressTimestampFormat = new ThreadLocal[SimpleDateFormat]() {
- override def initialValue(): SimpleDateFormat = {
- val format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'") // ISO8601
- format.setTimeZone(getTimeZone("UTC"))
- format
- }
- }
-
- def parseProgressTimestamp(timestamp: String): Long = {
- progressTimestampFormat.get.parse(timestamp).getTime
- }
-}
diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
index 130d22842b3..70db03df7bd 100644
--- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
+++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala
@@ -140,6 +140,7 @@ object CheckConnectJvmClientCompatibility {
ProblemFilters.exclude[Problem]("org.apache.spark.sql.internal.*"),
ProblemFilters.exclude[Problem]("org.apache.spark.sql.jdbc.*"),
ProblemFilters.exclude[Problem]("org.apache.spark.sql.sources.*"),
+ ProblemFilters.exclude[Problem]("org.apache.spark.sql.streaming.ui.*"),
ProblemFilters.exclude[Problem]("org.apache.spark.sql.test.*"),
ProblemFilters.exclude[Problem]("org.apache.spark.sql.util.*"),
@@ -280,24 +281,6 @@ object CheckConnectJvmClientCompatibility {
ProblemFilters.exclude[MissingClassProblem](
"org.apache.spark.sql.streaming.PythonStreamingQueryListenerWrapper"),
- // Streaming UI
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StreamingQueryDataSource"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StreamingQueryPagedTable"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StructuredStreamingRow"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StructuredStreamingRow$"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StreamingQueryPage"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StreamingQueryStatisticsPage"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StreamingQueryTab"),
- ProblemFilters.exclude[MissingClassProblem](
- "org.apache.spark.sql.streaming.ui.StreamingQueryTab$"),
-
// SQLImplicits
ProblemFilters.exclude[Problem]("org.apache.spark.sql.SQLImplicits.rddToDatasetHolder"),
ProblemFilters.exclude[Problem]("org.apache.spark.sql.SQLImplicits._sqlContext"))
diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
index b1fe94fbfbf..1287176d76e 100644
--- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
+++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
@@ -271,8 +271,7 @@ class StreamingQuerySuite extends QueryTest with SQLHelper with Logging {
}
test("streaming query listener") {
- // There's a default StreamingQueryStatusListener.
- assert(spark.streams.listListeners().length == 1)
+ assert(spark.streams.listListeners().length == 0)
val listener = new EventCollector
spark.streams.addListener(listener)
@@ -297,30 +296,25 @@ class StreamingQuerySuite extends QueryTest with SQLHelper with Logging {
// List listeners after adding a new listener, length should be 2.
val listeners = spark.streams.listListeners()
- assert(listeners.length == 2)
-
- val currListener = listeners(1).asInstanceOf[EventCollector]
- assert(q.id.equals(currListener.startEvent.id))
- assert(q.runId.equals(currListener.terminationEvent.runId))
- assert(q.lastProgress.numInputRows == currListener.progressEvents.last.numInputRows)
+ assert(listeners.length == 1)
// Add listener1 as another instance of EventCollector and validate
val listener1 = new EventCollector
spark.streams.addListener(listener1)
- assert(spark.streams.listListeners().length == 3)
- spark.streams.removeListener(listener1)
assert(spark.streams.listListeners().length == 2)
+ spark.streams.removeListener(listener1)
+ assert(spark.streams.listListeners().length == 1)
// Add the same listener again and validate, this aims to verify the listener cache
// is correctly stored and cleaned.
spark.streams.addListener(listener)
- assert(spark.streams.listListeners().length == 3)
- spark.streams.removeListener(listener)
assert(spark.streams.listListeners().length == 2)
+ spark.streams.removeListener(listener)
+ assert(spark.streams.listListeners().length == 1)
// Remove the listener, length should be 1.
spark.streams.removeListener(listener)
- assert(spark.streams.listListeners().length == 1)
+ assert(spark.streams.listListeners().length == 0)
}
test("foreachBatch") {
diff --git a/connector/connect/common/src/main/protobuf/spark/connect/commands.proto b/connector/connect/common/src/main/protobuf/spark/connect/commands.proto
index ce8c1d53943..4c4233124d8 100644
--- a/connector/connect/common/src/main/protobuf/spark/connect/commands.proto
+++ b/connector/connect/common/src/main/protobuf/spark/connect/commands.proto
@@ -400,7 +400,8 @@ message StreamingQueryManagerCommandResult {
}
message ListStreamingQueryListenerResult {
- repeated StreamingQueryListenerInstance listeners = 1;
+ // (Required) Reference IDs of listener instances.
+ repeated string listener_ids = 1;
}
}
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index d414169fd81..f82cb676040 100644
--- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -36,7 +36,7 @@ import org.apache.spark.connect.proto.Parse.ParseFormat
import org.apache.spark.connect.proto.StreamingForeachFunction
import org.apache.spark.connect.proto.StreamingQueryManagerCommand
import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult
-import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult.{StreamingQueryInstance, StreamingQueryListenerInstance}
+import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult.StreamingQueryInstance
import org.apache.spark.connect.proto.WriteStreamOperationStart.TriggerCase
import org.apache.spark.internal.Logging
import org.apache.spark.ml.{functions => MLFunctions}
@@ -2980,15 +2980,6 @@ class SparkConnectPlanner(val sessionHolder: SessionHolder) extends Logging {
builder.build()
}
- private def buildStreamingQueryListenerInstance(
- listener: StreamingQueryListener): StreamingQueryListenerInstance = {
- StreamingQueryListenerInstance
- .newBuilder()
- .setListenerPayload(ByteString
- .copyFrom(Utils.serialize(StreamingListenerPacket("", listener))))
- .build()
- }
-
def handleStreamingQueryManagerCommand(
command: StreamingQueryManagerCommand,
sessionId: String,
@@ -3047,12 +3038,8 @@ class SparkConnectPlanner(val sessionHolder: SessionHolder) extends Logging {
respBuilder.setRemoveListener(true)
case StreamingQueryManagerCommand.CommandCase.LIST_LISTENERS =>
- val listeners = session.streams.listListeners()
- respBuilder.getListListenersBuilder.addAllListeners(
- listeners
- .map(listener => buildStreamingQueryListenerInstance(listener))
- .toIterable
- .asJava)
+ respBuilder.getListListenersBuilder
+ .addAllListenerIds(sessionHolder.listListenerIds().asJava)
case StreamingQueryManagerCommand.CommandCase.COMMAND_NOT_SET =>
throw new IllegalArgumentException("Missing command in StreamingQueryManagerCommand")
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala
index 7361e370062..2f3bd1badce 100644
--- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala
@@ -184,6 +184,13 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
private[connect] def removeCachedListener(id: String): StreamingQueryListener = {
listenerCache.remove(id)
}
+
+ /**
+ * List listener IDs that have been register on server side.
+ */
+ private[connect] def listListenerIds(): Seq[String] = {
+ listenerCache.keySet().asScala.toSeq
+ }
}
object SessionHolder {
diff --git a/python/pyspark/sql/connect/proto/commands_pb2.py b/python/pyspark/sql/connect/proto/commands_pb2.py
index c852cebd140..6f03d80e669 100644
--- a/python/pyspark/sql/connect/proto/commands_pb2.py
+++ b/python/pyspark/sql/connect/proto/commands_pb2.py
@@ -35,7 +35,7 @@ from pyspark.sql.connect.proto import relations_pb2 as spark_dot_connect_dot_rel
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
- b'\n\x1cspark/connect/commands.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1aspark/connect/common.proto\x1a\x1fspark/connect/expressions.proto\x1a\x1dspark/connect/relations.proto"\xf5\x07\n\x07\x43ommand\x12]\n\x11register_function\x18\x01 \x01(\x0b\x32..spark.connect.CommonInlineUserDefinedFunctionH\x00R\x10registerFunction\x12H\n\x0fwrite_operation\x18\x02 \x01(\x0b\x32\x1d.spark.connect.WriteOperationH\x00R\x0ewriteOperation\x12_\n\x15\x63reate_dataframe_view\x [...]
+ b'\n\x1cspark/connect/commands.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1aspark/connect/common.proto\x1a\x1fspark/connect/expressions.proto\x1a\x1dspark/connect/relations.proto"\xf5\x07\n\x07\x43ommand\x12]\n\x11register_function\x18\x01 \x01(\x0b\x32..spark.connect.CommonInlineUserDefinedFunctionH\x00R\x10registerFunction\x12H\n\x0fwrite_operation\x18\x02 \x01(\x0b\x32\x1d.spark.connect.WriteOperationH\x00R\x0ewriteOperation\x12_\n\x15\x63reate_dataframe_view\x [...]
)
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
@@ -121,7 +121,7 @@ if _descriptor._USE_C_DESCRIPTORS == False:
_STREAMINGQUERYMANAGERCOMMAND_STREAMINGQUERYLISTENERCOMMAND._serialized_start = 7016
_STREAMINGQUERYMANAGERCOMMAND_STREAMINGQUERYLISTENERCOMMAND._serialized_end = 7090
_STREAMINGQUERYMANAGERCOMMANDRESULT._serialized_start = 7104
- _STREAMINGQUERYMANAGERCOMMANDRESULT._serialized_end = 8258
+ _STREAMINGQUERYMANAGERCOMMANDRESULT._serialized_end = 8180
_STREAMINGQUERYMANAGERCOMMANDRESULT_ACTIVERESULT._serialized_start = 7712
_STREAMINGQUERYMANAGERCOMMANDRESULT_ACTIVERESULT._serialized_end = 7839
_STREAMINGQUERYMANAGERCOMMANDRESULT_STREAMINGQUERYINSTANCE._serialized_start = 7841
@@ -130,12 +130,12 @@ if _descriptor._USE_C_DESCRIPTORS == False:
_STREAMINGQUERYMANAGERCOMMANDRESULT_AWAITANYTERMINATIONRESULT._serialized_end = 8017
_STREAMINGQUERYMANAGERCOMMANDRESULT_STREAMINGQUERYLISTENERINSTANCE._serialized_start = 8019
_STREAMINGQUERYMANAGERCOMMANDRESULT_STREAMINGQUERYLISTENERINSTANCE._serialized_end = 8094
- _STREAMINGQUERYMANAGERCOMMANDRESULT_LISTSTREAMINGQUERYLISTENERRESULT._serialized_start = 8097
- _STREAMINGQUERYMANAGERCOMMANDRESULT_LISTSTREAMINGQUERYLISTENERRESULT._serialized_end = 8243
- _GETRESOURCESCOMMAND._serialized_start = 8260
- _GETRESOURCESCOMMAND._serialized_end = 8281
- _GETRESOURCESCOMMANDRESULT._serialized_start = 8284
- _GETRESOURCESCOMMANDRESULT._serialized_end = 8496
- _GETRESOURCESCOMMANDRESULT_RESOURCESENTRY._serialized_start = 8400
- _GETRESOURCESCOMMANDRESULT_RESOURCESENTRY._serialized_end = 8496
+ _STREAMINGQUERYMANAGERCOMMANDRESULT_LISTSTREAMINGQUERYLISTENERRESULT._serialized_start = 8096
+ _STREAMINGQUERYMANAGERCOMMANDRESULT_LISTSTREAMINGQUERYLISTENERRESULT._serialized_end = 8165
+ _GETRESOURCESCOMMAND._serialized_start = 8182
+ _GETRESOURCESCOMMAND._serialized_end = 8203
+ _GETRESOURCESCOMMANDRESULT._serialized_start = 8206
+ _GETRESOURCESCOMMANDRESULT._serialized_end = 8418
+ _GETRESOURCESCOMMANDRESULT_RESOURCESENTRY._serialized_start = 8322
+ _GETRESOURCESCOMMANDRESULT_RESOURCESENTRY._serialized_end = 8418
# @@protoc_insertion_point(module_scope)
diff --git a/python/pyspark/sql/connect/proto/commands_pb2.pyi b/python/pyspark/sql/connect/proto/commands_pb2.pyi
index 5b44e3da7a0..be423ea036e 100644
--- a/python/pyspark/sql/connect/proto/commands_pb2.pyi
+++ b/python/pyspark/sql/connect/proto/commands_pb2.pyi
@@ -1566,23 +1566,19 @@ class StreamingQueryManagerCommandResult(google.protobuf.message.Message):
class ListStreamingQueryListenerResult(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.Descriptor
- LISTENERS_FIELD_NUMBER: builtins.int
+ LISTENER_IDS_FIELD_NUMBER: builtins.int
@property
- def listeners(
+ def listener_ids(
self,
- ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
- global___StreamingQueryManagerCommandResult.StreamingQueryListenerInstance
- ]: ...
+ ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]:
+ """(Required) Reference IDs of listener instances."""
def __init__(
self,
*,
- listeners: collections.abc.Iterable[
- global___StreamingQueryManagerCommandResult.StreamingQueryListenerInstance
- ]
- | None = ...,
+ listener_ids: collections.abc.Iterable[builtins.str] | None = ...,
) -> None: ...
def ClearField(
- self, field_name: typing_extensions.Literal["listeners", b"listeners"]
+ self, field_name: typing_extensions.Literal["listener_ids", b"listener_ids"]
) -> None: ...
ACTIVE_FIELD_NUMBER: builtins.int
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
index 1b4c8dbd5c8..aae2fad780c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
@@ -38,8 +38,8 @@ import org.apache.spark.util.kvstore.KVIndex
* UI data for both active and inactive query.
*/
private[sql] class StreamingQueryStatusListener(
- @transient val conf: SparkConf,
- @transient val store: ElementTrackingStore) extends StreamingQueryListener {
+ val conf: SparkConf,
+ val store: ElementTrackingStore) extends StreamingQueryListener {
private val streamingProgressRetention =
conf.get(StaticSQLConf.STREAMING_UI_RETAINED_PROGRESS_UPDATES)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org