[SPARK-58369][CONNECT][TESTS] Introduce a way to SparkConnectServiceKeepAliveSuite to wait until unbinding a port is complete

[sarutak]

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() in Linux.
doClose() is also called from AbstractChannel#doClose0()
In Linux environment, doClose0() is called asynchronously here because EpollSocketChannelUnsafe overrides prepareToClose() and closeExecutor 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.

[uros-b]

I think that this might be duplicating an existing draft PR (#57436) - cc @Ma77Ball who tried to fix the same flake via OS-assigned ports. Let's consolidate on one approach

[sarutak]

Both PRs take a different approach. Personally, I prefer this one because it's simpler, but I'll leave it up to the reviewers to decide which solution to adopt.

[dongjoon-hyun]

+1, LGTM.