HyukjinKwon commented on code in PR #46819:
URL: https://github.com/apache/spark/pull/46819#discussion_r1623387301
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -179,12 +179,13 @@ case class SessionHolder(userId: String, sessionId:
String, session: SparkSessio
*/
private[service] def interruptAll(): Seq[String] = {
val interruptedIds = new mutable.ArrayBuffer[String]()
+ val operationsIds =
SparkConnectService.streamingSessionManager.cleanupRunningQueries(this)
executions.asScala.values.foreach { execute =>
if (execute.interrupt()) {
interruptedIds += execute.operationId
}
}
- interruptedIds.toSeq
+ interruptedIds.toSeq ++ operationsIds
Review Comment:
This is a little bit weird one. The operations IDs are not from actual
running `executions` within `SessionHolder` but the operation ID from the past
execution (that launched a streaming query).
I am not sure if it makes sense, or if we should do something else,
@juliuszsompolski @WweiL
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
