You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "WweiL (via GitHub)" <gi...@apache.org> on 2023/05/03 23:00:35 UTC
[GitHub] [spark] WweiL opened a new pull request, #41039: [SPARK-43360] [SS] [CONNECT] Scala client StreamingQueryManager
WweiL opened a new pull request, #41039:
URL: https://github.com/apache/spark/pull/41039
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Add support for scala client `StreamingQueryManager`
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Development of scala connect client
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes they can use SQM in scala client
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Manually tested
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1193352398
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQuery.scala:
##########
@@ -297,4 +298,20 @@ object RemoteStreamingQuery {
name = if (result.getName.isEmpty) null else result.getName,
sparkSession = sparkSession)
}
+
+ def fromStreamingQueryInstanceResponse(
+ sparkSession: SparkSession,
+ q: StreamingQueryInstance): RemoteStreamingQuery = {
+
+ val name = if (q.hasName) {
+ q.getName
+ } else {
+ null
Review Comment:
Yes, also the original `StreamingQuery`. This is to align the behavior
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1193425714
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala:
##########
@@ -131,6 +131,7 @@ object CheckConnectJvmClientCompatibility {
IncludeByName("org.apache.spark.sql.streaming.DataStreamReader.*"),
IncludeByName("org.apache.spark.sql.streaming.DataStreamWriter.*"),
IncludeByName("org.apache.spark.sql.streaming.StreamingQuery.*"),
+ IncludeByName("org.apache.spark.sql.streaming.StreamingQueryManager.*"),
Review Comment:
Too many shouldn't support methods if doing `IncludeByName("org.apache.spark.sql.streaming.StreamingQueryManager.*"),`
:
```
method logName()java.lang.String in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method log()org.slf4j.Logger in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logInfo(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logDebug(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logTrace(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logWarning(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logError(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logInfo(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logDebug(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logTrace(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logWarning(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logError(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method isTraceEnabled()Boolean in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method initializeLogIfNecessary(Boolean)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method initializeLogIfNecessary(Boolean,Boolean)Boolean in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
synthetic method initializeLogIfNecessary$default$2()Boolean in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method initializeForcefully(Boolean,Boolean)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method addListener(org.apache.spark.sql.streaming.StreamingQueryListener)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method removeListener(org.apache.spark.sql.streaming.StreamingQueryListener)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method listListeners()Array[org.apache.spark.sql.streaming.StreamingQueryListener] in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method this(org.apache.spark.sql.SparkSession,org.apache.spark.sql.internal.SQLConf)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
Exceptions to binary compatibility can be added in 'CheckConnectJvmClientCompatibility#checkMiMaCompatibility'
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1194437164
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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
+
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.connect.proto.Command
+import org.apache.spark.connect.proto.StreamingQueryManagerCommand
+import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult
+import org.apache.spark.sql.SparkSession
+
+/**
+ * A class to manage all the [[StreamingQuery]] active in a `SparkSession`.
+ *
+ * @since 3.5.0
+ */
+@Evolving
+class StreamingQueryManager private[sql] (sparkSession: SparkSession) {
+
+ /**
+ * Returns a list of active queries associated with this SQLContext
+ *
+ * @since 3.5.0
+ */
+ def active: Array[StreamingQuery] = {
+ executeManagerCmd(_.setActive(true)).getActive.getActiveQueriesList.asScala.map { q =>
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, q)
+ }.toArray
+ }
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: UUID): StreamingQuery = get(id.toString)
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: String): StreamingQuery = {
+ val response = executeManagerCmd(_.setGet(id))
+ if (response.hasQuery) {
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, response.getQuery)
+ } else {
+ null
+ }
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. If any query was terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return immediately (if the query was terminated by `query.stop()`), or throw the exception
+ * immediately (if the query was terminated with exception). Use `resetTerminated()` to clear
+ * past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(): Unit = {
+ executeManagerCmd(_.getAwaitAnyTerminationBuilder.build())
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. Returns whether any query has
+ * terminated or not (multiple may have terminated). If any query has terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return `true` immediately (if the query was terminated by `query.stop()`), or throw the
+ * exception immediately (if the query was terminated with exception). Use `resetTerminated()`
+ * to clear past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(timeoutMs: Long): Boolean = {
+ require(timeoutMs > 0, "Timeout has to be positive")
+ executeManagerCmd(
+ _.getAwaitAnyTerminationBuilder.setTimeoutMs(
+ timeoutMs)).getAwaitAnyTermination.getTerminated
+ }
+
+ /**
+ * Forget about past terminated queries so that `awaitAnyTermination()` can be used again to
+ * wait for new terminations.
+ *
+ * @since 3.5.0
+ */
+ def resetTerminated(): Unit = {
+ executeManagerCmd(_.setResetTerminated(true))
+ }
+
+ private def executeManagerCmd(
Review Comment:
OK, sg.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1192857282
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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
+
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.connect.proto.Command
+import org.apache.spark.connect.proto.StreamingQueryManagerCommand
+import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult
+import org.apache.spark.sql.SparkSession
+
+/**
+ * A class to manage all the [[StreamingQuery]] active in a `SparkSession`.
+ *
+ * @since 3.5.0
+ */
+@Evolving
+class StreamingQueryManager private[sql] (sparkSession: SparkSession) {
+
+ /**
+ * Returns a list of active queries associated with this SQLContext
+ *
+ * @since 3.5.0
+ */
+ def active: Array[StreamingQuery] = {
+ executeManagerCmd(_.setActive(true)).getActive.getActiveQueriesList.asScala.map { q =>
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, q)
+ }.toArray
+ }
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: UUID): StreamingQuery = get(id.toString)
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: String): StreamingQuery = {
+ val response = executeManagerCmd(_.setGet(id))
+ if (response.hasQuery) {
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, response.getQuery)
+ } else {
+ null
+ }
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. If any query was terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return immediately (if the query was terminated by `query.stop()`), or throw the exception
+ * immediately (if the query was terminated with exception). Use `resetTerminated()` to clear
+ * past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(): Unit = {
+ executeManagerCmd(_.getAwaitAnyTerminationBuilder.build())
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. Returns whether any query has
+ * terminated or not (multiple may have terminated). If any query has terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return `true` immediately (if the query was terminated by `query.stop()`), or throw the
+ * exception immediately (if the query was terminated with exception). Use `resetTerminated()`
+ * to clear past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(timeoutMs: Long): Boolean = {
+ require(timeoutMs > 0, "Timeout has to be positive")
+ executeManagerCmd(
+ _.getAwaitAnyTerminationBuilder.setTimeoutMs(
+ timeoutMs)).getAwaitAnyTermination.getTerminated
+ }
+
+ /**
+ * Forget about past terminated queries so that `awaitAnyTermination()` can be used again to
+ * wait for new terminations.
+ *
+ * @since 3.5.0
+ */
+ def resetTerminated(): Unit = {
+ executeManagerCmd(_.setResetTerminated(true))
+ }
+
+ private def executeManagerCmd(
Review Comment:
Not a blocker: this method looks very similar to `executeQueryCmd` in `StreamingQuery`. Do you think if there's an easy way to abstract that logic out as a common/util method that could be shared among different commands? If no easy solution is available, you can ignore this comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1193351864
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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
+
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.connect.proto.Command
+import org.apache.spark.connect.proto.StreamingQueryManagerCommand
+import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult
+import org.apache.spark.sql.SparkSession
+
+/**
+ * A class to manage all the [[StreamingQuery]] active in a `SparkSession`.
+ *
+ * @since 3.5.0
+ */
+@Evolving
+class StreamingQueryManager private[sql] (sparkSession: SparkSession) {
+
+ /**
+ * Returns a list of active queries associated with this SQLContext
+ *
+ * @since 3.5.0
+ */
+ def active: Array[StreamingQuery] = {
+ executeManagerCmd(_.setActive(true)).getActive.getActiveQueriesList.asScala.map { q =>
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, q)
+ }.toArray
+ }
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: UUID): StreamingQuery = get(id.toString)
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: String): StreamingQuery = {
+ val response = executeManagerCmd(_.setGet(id))
+ if (response.hasQuery) {
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, response.getQuery)
+ } else {
+ null
+ }
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. If any query was terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return immediately (if the query was terminated by `query.stop()`), or throw the exception
+ * immediately (if the query was terminated with exception). Use `resetTerminated()` to clear
+ * past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(): Unit = {
+ executeManagerCmd(_.getAwaitAnyTerminationBuilder.build())
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. Returns whether any query has
+ * terminated or not (multiple may have terminated). If any query has terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return `true` immediately (if the query was terminated by `query.stop()`), or throw the
+ * exception immediately (if the query was terminated with exception). Use `resetTerminated()`
+ * to clear past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(timeoutMs: Long): Boolean = {
+ require(timeoutMs > 0, "Timeout has to be positive")
+ executeManagerCmd(
+ _.getAwaitAnyTerminationBuilder.setTimeoutMs(
+ timeoutMs)).getAwaitAnyTermination.getTerminated
+ }
+
+ /**
+ * Forget about past terminated queries so that `awaitAnyTermination()` can be used again to
+ * wait for new terminations.
+ *
+ * @since 3.5.0
+ */
+ def resetTerminated(): Unit = {
+ executeManagerCmd(_.setResetTerminated(true))
+ }
+
+ private def executeManagerCmd(
Review Comment:
I think this is possible but not necessarily a urgent need as we only have two of these right now. Also these methods :`getStreamingQueryManagerCommandBuilder` and `getStreamingQueryCommandBuilder` and other similar methods makes it a little bit to generalize I'd say...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1193425714
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala:
##########
@@ -131,6 +131,7 @@ object CheckConnectJvmClientCompatibility {
IncludeByName("org.apache.spark.sql.streaming.DataStreamReader.*"),
IncludeByName("org.apache.spark.sql.streaming.DataStreamWriter.*"),
IncludeByName("org.apache.spark.sql.streaming.StreamingQuery.*"),
+ IncludeByName("org.apache.spark.sql.streaming.StreamingQueryManager.*"),
Review Comment:
Too many shouldn't supported methods got reported if doing `IncludeByName("org.apache.spark.sql.streaming.StreamingQueryManager.*"),`
:
```
method logName()java.lang.String in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method log()org.slf4j.Logger in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logInfo(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logDebug(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logTrace(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logWarning(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logError(scala.Function0)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logInfo(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logDebug(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logTrace(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logWarning(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method logError(scala.Function0,java.lang.Throwable)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method isTraceEnabled()Boolean in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method initializeLogIfNecessary(Boolean)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method initializeLogIfNecessary(Boolean,Boolean)Boolean in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
synthetic method initializeLogIfNecessary$default$2()Boolean in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method initializeForcefully(Boolean,Boolean)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method addListener(org.apache.spark.sql.streaming.StreamingQueryListener)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method removeListener(org.apache.spark.sql.streaming.StreamingQueryListener)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method listListeners()Array[org.apache.spark.sql.streaming.StreamingQueryListener] in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
method this(org.apache.spark.sql.SparkSession,org.apache.spark.sql.internal.SQLConf)Unit in class org.apache.spark.sql.streaming.StreamingQueryManager does not have a correspondent in client version
Exceptions to binary compatibility can be added in 'CheckConnectJvmClientCompatibility#checkMiMaCompatibility'
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon closed pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
URL: https://github.com/apache/spark/pull/41039
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on PR #41039:
URL: https://github.com/apache/spark/pull/41039#issuecomment-1543167284
@WweiL can you add `StreamingManager` to the `CheckConnectJvmClientCompatibility` test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rangadi commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1193360679
##########
connector/connect/common/src/main/protobuf/spark/connect/commands.proto:
##########
@@ -360,7 +360,7 @@ message StreamingQueryManagerCommandResult {
// (Required) The id and runId of this query.
StreamingQueryInstanceId id = 1;
// (Optional) The name of this query.
- string name = 2;
+ optional string name = 2;
Review Comment:
I think so, yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on pull request #41039: [SPARK-43360] [SS] [CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on PR #41039:
URL: https://github.com/apache/spark/pull/41039#issuecomment-1533860277
@rangadi @pengzhon-db
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rangadi commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1190357553
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQuery.scala:
##########
@@ -297,4 +298,15 @@ object RemoteStreamingQuery {
name = if (result.getName.isEmpty) null else result.getName,
sparkSession = sparkSession)
}
+
+ def fromStreamingQueryInstanceResponse(
+ sparkSession: SparkSession,
+ q: StreamingQueryInstance): RemoteStreamingQuery = {
+
+ new RemoteStreamingQuery(
+ UUID.fromString(q.getId.getId),
+ UUID.fromString(q.getId.getRunId),
+ q.getName,
Review Comment:
Should this be null if name is not set? I don't think empty string is correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rangadi commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1191748881
##########
connector/connect/common/src/main/protobuf/spark/connect/commands.proto:
##########
@@ -360,7 +360,7 @@ message StreamingQueryManagerCommandResult {
// (Required) The id and runId of this query.
StreamingQueryInstanceId id = 1;
// (Optional) The name of this query.
- string name = 2;
+ optional string name = 2;
Review Comment:
We can also check if the value empty and treat it as missing. Not a fan of heavy use of optional keyword (Proto3 reluctantly supports it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1192902046
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQuery.scala:
##########
@@ -297,4 +298,20 @@ object RemoteStreamingQuery {
name = if (result.getName.isEmpty) null else result.getName,
sparkSession = sparkSession)
}
+
+ def fromStreamingQueryInstanceResponse(
+ sparkSession: SparkSession,
+ q: StreamingQueryInstance): RemoteStreamingQuery = {
+
+ val name = if (q.hasName) {
+ q.getName
+ } else {
+ null
Review Comment:
Oh because `RemoteStreamingQuery` was not designed to take Option?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on pull request #41039: [SPARK-43360] [SS] [CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on PR #41039:
URL: https://github.com/apache/spark/pull/41039#issuecomment-1533860158
Some of the bug fixes are also in https://github.com/apache/spark/pull/41037
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1191749684
##########
connector/connect/common/src/main/protobuf/spark/connect/commands.proto:
##########
@@ -360,7 +360,7 @@ message StreamingQueryManagerCommandResult {
// (Required) The id and runId of this query.
StreamingQueryInstanceId id = 1;
// (Optional) The name of this query.
- string name = 2;
+ optional string name = 2;
Review Comment:
I see. But then if a user deliberately set the query name to be `""`, it would be treated as null also. We are fine with that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1191747787
##########
connector/connect/common/src/main/protobuf/spark/connect/commands.proto:
##########
@@ -360,7 +360,7 @@ message StreamingQueryManagerCommandResult {
// (Required) The id and runId of this query.
StreamingQueryInstanceId id = 1;
// (Optional) The name of this query.
- string name = 2;
+ optional string name = 2;
Review Comment:
`optional` is needed to use the `hasName` method in scala
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1192901988
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQuery.scala:
##########
@@ -297,4 +298,20 @@ object RemoteStreamingQuery {
name = if (result.getName.isEmpty) null else result.getName,
sparkSession = sparkSession)
}
+
+ def fromStreamingQueryInstanceResponse(
+ sparkSession: SparkSession,
+ q: StreamingQueryInstance): RemoteStreamingQuery = {
+
+ val name = if (q.hasName) {
+ q.getName
+ } else {
+ null
Review Comment:
Why null but not `Option`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #41039:
URL: https://github.com/apache/spark/pull/41039#discussion_r1193351864
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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
+
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.connect.proto.Command
+import org.apache.spark.connect.proto.StreamingQueryManagerCommand
+import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult
+import org.apache.spark.sql.SparkSession
+
+/**
+ * A class to manage all the [[StreamingQuery]] active in a `SparkSession`.
+ *
+ * @since 3.5.0
+ */
+@Evolving
+class StreamingQueryManager private[sql] (sparkSession: SparkSession) {
+
+ /**
+ * Returns a list of active queries associated with this SQLContext
+ *
+ * @since 3.5.0
+ */
+ def active: Array[StreamingQuery] = {
+ executeManagerCmd(_.setActive(true)).getActive.getActiveQueriesList.asScala.map { q =>
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, q)
+ }.toArray
+ }
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: UUID): StreamingQuery = get(id.toString)
+
+ /**
+ * Returns the query if there is an active query with the given id, or null.
+ *
+ * @since 3.5.0
+ */
+ def get(id: String): StreamingQuery = {
+ val response = executeManagerCmd(_.setGet(id))
+ if (response.hasQuery) {
+ RemoteStreamingQuery.fromStreamingQueryInstanceResponse(sparkSession, response.getQuery)
+ } else {
+ null
+ }
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. If any query was terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return immediately (if the query was terminated by `query.stop()`), or throw the exception
+ * immediately (if the query was terminated with exception). Use `resetTerminated()` to clear
+ * past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(): Unit = {
+ executeManagerCmd(_.getAwaitAnyTerminationBuilder.build())
+ }
+
+ /**
+ * Wait until any of the queries on the associated SQLContext has terminated since the creation
+ * of the context, or since `resetTerminated()` was called. Returns whether any query has
+ * terminated or not (multiple may have terminated). If any query has terminated with an
+ * exception, then the exception will be thrown.
+ *
+ * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
+ * return `true` immediately (if the query was terminated by `query.stop()`), or throw the
+ * exception immediately (if the query was terminated with exception). Use `resetTerminated()`
+ * to clear past terminations and wait for new terminations.
+ *
+ * For correctly documenting exceptions across multiple queries, users need to stop all of them
+ * after any of them terminates with exception, and then check the `query.exception()` for each
+ * query.
+ *
+ * @since 3.5.0
+ */
+ // TODO(SPARK-43299): verity the behavior of this method after JVM client-side error-handling
+ // framework is supported and modify the doc accordingly.
+ def awaitAnyTermination(timeoutMs: Long): Boolean = {
+ require(timeoutMs > 0, "Timeout has to be positive")
+ executeManagerCmd(
+ _.getAwaitAnyTerminationBuilder.setTimeoutMs(
+ timeoutMs)).getAwaitAnyTermination.getTerminated
+ }
+
+ /**
+ * Forget about past terminated queries so that `awaitAnyTermination()` can be used again to
+ * wait for new terminations.
+ *
+ * @since 3.5.0
+ */
+ def resetTerminated(): Unit = {
+ executeManagerCmd(_.setResetTerminated(true))
+ }
+
+ private def executeManagerCmd(
Review Comment:
I think this is possible but not necessarily a urgent need as we only have two of these right now. Also these methods :`getStreamingQueryManagerCommandBuilder` and `getStreamingQueryCommandBuilder` and other similar methods makes it a little bit to generally I'd say...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #41039: [SPARK-43360][SS][CONNECT] Scala client StreamingQueryManager
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #41039:
URL: https://github.com/apache/spark/pull/41039#issuecomment-1550483880
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org