sarutak opened a new pull request, #57560: URL: https://github.com/apache/spark/pull/57560
### What changes were proposed in this pull request? This PR introduce `stopSparkConnectServiceAndUnbind()` to `SparkConnectServiceKeepAliveSuite` as a way to wait until unbinding a port is complete. ### Why are the changes needed? SparkConnectServiceKeepAliveSuite sometimes flakily fails. #57342 tried to fix it but it but it still happens. https://github.com/apache/spark/actions/runs/30041141074/job/89325300649 ``` [info] - SPARK-58094: disabling spark.connect.grpc.keepAlive.enabled reverts to the pre-fix hang *** FAILED *** (6 milliseconds) [info] java.net.BindException: Failed to bind to address 0.0.0.0/0.0.0.0:15788: Service 'org.apache.spark.sql.connect.service.SparkConnectService' failed after 0 retries (starting from 15788)! Consider explicitly setting the appropriate port for the service 'org.apache.spark.sql.connect.service.SparkConnectService' (for example spark.ui.port for SparkUI) to an available port or increasing spark.port.maxRetries. [info] at io.grpc.netty.NettyServer.start(NettyServer.java:341) [info] at io.grpc.internal.ServerImpl.start(ServerImpl.java:185) [info] at io.grpc.internal.ServerImpl.start(ServerImpl.java:94) [info] at org.apache.spark.sql.connect.service.SparkConnectService$.$anonfun$startGRPCService$1(SparkConnectService.scala:451) [info] at org.apache.spark.sql.connect.service.SparkConnectService$.$anonfun$startGRPCService$1$adapted(SparkConnectService.scala:411) [info] at org.apache.spark.util.Utils$.$anonfun$startServiceOnPort$2(Utils.scala:2276) [info] at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:256) [info] at org.apache.spark.util.Utils$.startServiceOnPort(Utils.scala:2268) [info] at org.apache.spark.sql.connect.service.SparkConnectService$.startGRPCService(SparkConnectService.scala:471) [info] at org.apache.spark.sql.connect.service.SparkConnectService$.start(SparkConnectService.scala:483) [info] at org.apache.spark.sql.connect.service.SparkConnectServiceKeepAliveSuite.$anonfun$new$17(SparkConnectServiceKeepAliveSuite.scala:262) [info] at org.apache.spark.SparkTestSuite.withSparkEnvConfs(SparkTestSuite.scala:268) [info] at org.apache.spark.SparkTestSuite.withSparkEnvConfs$(SparkTestSuite.scala:257) [info] at org.apache.spark.SparkFunSuite.withSparkEnvConfs(SparkFunSuite.scala:33) [info] at org.apache.spark.sql.connect.service.SparkConnectServiceKeepAliveSuite.$anonfun$new$16(SparkConnectServiceKeepAliveSuite.scala:262) ... ``` The root cause of the issue is that `SparkConnectService.stop()` may return before unbinding the port is complete in Linux environment even though `io.grpc.Server#awaitTermination()` is called within `SparkConnectService.stop()`. We can see this behavior with the following code. With this code, `SparkConnectService.start()` may throw `BindException`. ``` for (_ <- 1 to 3000) { SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) withSparkEnvConfs( Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { SparkConnectService.start(spark.sparkContext) } } ``` We use Netty in `SparkConnectService` so the socket is closed in `AbstractChannel#doClose()`, more specifically in [AbstractEpollChannel#doClose](https://github.com/netty/netty/blob/3703d79669ee024f6483c9d8697ac58ba546df33/transport-classes-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java#L214) in Linux. `doClose()` is also called from [AbstractChannel#doClose0()](https://github.com/netty/netty/blob/3703d79669ee024f6483c9d8697ac58ba546df33/transport/src/main/java/io/netty/channel/AbstractChannel.java#L617) In Linux environment, `doClose0()` is called asynchronously [here](https://github.com/netty/netty/blob/3703d79669ee024f6483c9d8697ac58ba546df33/transport/src/main/java/io/netty/channel/AbstractChannel.java#L574) because `EpollSocketChannelUnsafe` overrides [prepareToClos()](https://github.com/netty/netty/blob/3703d79669ee024f6483c9d8697ac58ba546df33/transport-classes-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java#L159) and [closeExecutor](https://github.com/netty/netty/blob/3703d79669ee024f6483c9d8697ac58ba546df33/transport/src/main/java/io/netty/channel/AbstractChannel.java#L567) will be non-null. On the other hand, I don't see `prepareToClose()` overridden in any class under `io.netty.channel.kqueue`. So the flakiness should not affect in Mac environment. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Confirmed that the following code doesn't throw `BindException`. ``` for (_ <- 1 to 3000) { stopSparkConnectServiceAndUnbind() withSparkEnvConfs( Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { SparkConnectService.start(spark.sparkContext) } } ``` ### Was this patch authored or co-authored using generative AI tooling? No. -- 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]
