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]

Reply via email to