[
https://issues.apache.org/jira/browse/SPARK-35154?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
LIU updated SPARK-35154:
------------------------
Description:
when i use this code to work, Rpc thread hangs up and not close gracefully. i
think when rpc thread called shutdown on OnStop method, it will try to put
MessageLoop.PoisonPill to return and stop thread in rpc pool. In spark 3.x, it
will make others thread return & stop but current thread which call OnStop
method to await current pool to stop. it makes current thread not stop, and
pending program.
I'm not sure that needs to be improved or not?
{code:java}
//代码占位符{code}
test("Rpc env not shutdown when shutdown method call by endpoint onStop") {
val rpcEndpoint = new RpcEndpoint {
override val rpcEnv: RpcEnv = env
override def onStop(): Unit = {
env.shutdown()
env.awaitTermination()
}
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any,
Unit] = {
case m => context.reply(m)
}
}
env.setupEndpoint("test", rpcEndpoint)
rpcEndpoint.stop()
env.awaitTermination()
}
> Rpc env not shutdown when shutdown method call by endpoint onStop
> -----------------------------------------------------------------
>
> Key: SPARK-35154
> URL: https://issues.apache.org/jira/browse/SPARK-35154
> Project: Spark
> Issue Type: Improvement
> Components: Spark Core
> Affects Versions: 3.0.0
> Environment: spark-3.x
> Reporter: LIU
> Priority: Minor
>
> when i use this code to work, Rpc thread hangs up and not close gracefully.
> i think when rpc thread called shutdown on OnStop method, it will try to put
> MessageLoop.PoisonPill to return and stop thread in rpc pool. In spark 3.x,
> it will make others thread return & stop but current thread which call OnStop
> method to await current pool to stop. it makes current thread not stop, and
> pending program.
> I'm not sure that needs to be improved or not?
>
> {code:java}
> //代码占位符{code}
> test("Rpc env not shutdown when shutdown method call by endpoint onStop") {
> val rpcEndpoint = new RpcEndpoint {
> override val rpcEnv: RpcEnv = env
> override def onStop(): Unit = {
> env.shutdown()
> env.awaitTermination()
> }
> override def receiveAndReply(context: RpcCallContext):
> PartialFunction[Any, Unit] = {
> case m => context.reply(m)
> }
> }
> env.setupEndpoint("test", rpcEndpoint)
> rpcEndpoint.stop()
> env.awaitTermination()
> }
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
