Re: [PR] [SPARK-48485][CONNECT][SS] Support interruptTag and interruptAll in streaming queries [spark]

["HyukjinKwon (via GitHub)" <gi...@apache.org>]

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: 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