ulysses-you commented on code in PR #4545:
URL: https://github.com/apache/kyuubi/pull/4545#discussion_r1142859745
##########
kyuubi-common/src/main/scala/org/apache/kyuubi/operation/OperationManager.scala:
##########
@@ -189,4 +192,30 @@ abstract class OperationManager(name: String) extends
AbstractService(name) {
null
}
}
+
+ def waitForOperationsToFinish(): Unit = {
+ val timeWhenStopStarted = System.nanoTime()
+ val stopTimeoutMs = conf.get(SESSION_CLOSE_GRACEFULLY_TIMEOUT)
+ val pollTime = 300
+
+ // To prevent graceful stop to get stuck permanently
+ def hasTimedOut: Boolean = {
+ val diff = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
timeWhenStopStarted)
+ val timedOut = diff > stopTimeoutMs
+ if (timedOut) {
+ warn("Timed out while waiting the spark operations to finish " +
+ "(timeout = " + stopTimeoutMs + ")")
+ }
+ timedOut
+ }
+
+ info("Waiting for all the operations to finish.")
+ while (!hasTimedOut && !allOperations().filter(op =>
Review Comment:
`allOperations().exists(...)` to avoid transform tail operations if has
active operation.
--
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]
