fix flaky test, clean up netty warning logs

Author: danielpeach

java-plugin/src/main/kotlin/io/github/danielpeach/plugin/Broker.kt

@@ -72,8 +72,6 @@ class Broker internal constructor(
    * @param [services] The gRPC services that the [Broker] should serve.
    *
    * @returns [Server] A gRPC server that implements the provided [services].
-   *
-   * @throws [BrokerServiceException]
    * */
   fun acceptAndServe(serviceId: Int, vararg services: BindableService): Server = runBlocking(scope.coroutineContext) {
     val server = channelProvider.server("unix", *services).start()
@@ -112,7 +110,7 @@ class Broker internal constructor(
       stub
         .startStream(sender.consumeAsFlow())
         .catch { e ->
-          sender.cancel(BrokerServiceException(e))
+          sender.cancel(CancellationException(e))
         }
         .collect { connection ->
           logger.debug("Received connection info for service #${connection.serviceId}")

java-plugin/src/main/kotlin/io/github/danielpeach/plugin/BrokerServiceException.kt

@@ -6,6 +6,7 @@ import io.grpc.Server
 import io.grpc.netty.GrpcSslContexts
 import io.grpc.netty.NettyChannelBuilder
 import io.grpc.netty.NettyServerBuilder
+import io.netty.channel.ChannelOption
 import io.netty.channel.epoll.EpollDomainSocketChannel
 import io.netty.channel.epoll.EpollEventLoopGroup
 import io.netty.channel.epoll.EpollServerDomainSocketChannel
@@ -36,24 +37,27 @@ internal class ChannelProvider(private val mTLSConfig: MTLSConfig? = null) {
       throw IllegalArgumentException("Network type '$network' is not supported.")
     }
 
-    val (builder, shutdownHook) = when (val os = System.getProperty("os.name")) {
+    val builder = NettyChannelBuilder.forAddress(DomainSocketAddress(address))
+      .withOption(ChannelOption.SO_KEEPALIVE, null)
+
+    val shutdownHook = when (val os = System.getProperty("os.name")) {
       MAC_OS -> {
         val kqg = KQueueEventLoopGroup()
-        val builder =
-          NettyChannelBuilder.forAddress(DomainSocketAddress(address))
-            .eventLoopGroup(kqg)
-            .channelType(KQueueDomainSocketChannel::class.java)
 
-        builder to { kqg.shutdownGracefully() }
+        builder
+          .eventLoopGroup(kqg)
+          .channelType(KQueueDomainSocketChannel::class.java);
+
+        { kqg.shutdownGracefully() }
       }
       LINUX -> {
         val elg = EpollEventLoopGroup()
-        val builder =
-          NettyChannelBuilder.forAddress(DomainSocketAddress(address))
-            .eventLoopGroup(elg)
-            .channelType(EpollDomainSocketChannel::class.java)
 
-        builder to { elg.shutdownGracefully() }
+        builder
+          .eventLoopGroup(elg)
+          .channelType(EpollDomainSocketChannel::class.java);
+
+        { elg.shutdownGracefully() }
       }
       else -> throw IllegalArgumentException("OS '$os' is not supported.")
     }
@@ -103,18 +107,20 @@ internal class ChannelProvider(private val mTLSConfig: MTLSConfig? = null) {
       throw IllegalArgumentException("Network type '$network' is not supported.")
     }
 
-    val (builder, shutdownHook) = when (val os = System.getProperty("os.name")) {
+    val builder = NettyServerBuilder.forAddress(DomainSocketAddress(address))
+      .withChildOption(ChannelOption.SO_KEEPALIVE, null)
+
+    val shutdownHook = when (val os = System.getProperty("os.name")) {
       MAC_OS -> {
         val boss = KQueueEventLoopGroup()
         val worker = KQueueEventLoopGroup()
 
-        val builder =
-          NettyServerBuilder.forAddress(DomainSocketAddress(address))
-            .bossEventLoopGroup(boss)
-            .workerEventLoopGroup(worker)
-            .channelType(KQueueServerDomainSocketChannel::class.java)
+        builder
+          .bossEventLoopGroup(boss)
+          .workerEventLoopGroup(worker)
+          .channelType(KQueueServerDomainSocketChannel::class.java);
 
-        builder to {
+        {
           boss.shutdownGracefully()
           worker.shutdownGracefully()
         }
@@ -123,13 +129,12 @@ internal class ChannelProvider(private val mTLSConfig: MTLSConfig? = null) {
         val boss = EpollEventLoopGroup()
         val worker = EpollEventLoopGroup()
 
-        val builder =
-          NettyServerBuilder.forAddress(DomainSocketAddress(address))
-            .bossEventLoopGroup(boss)
-            .workerEventLoopGroup(worker)
-            .channelType(EpollServerDomainSocketChannel::class.java)
+        builder
+          .bossEventLoopGroup(boss)
+          .workerEventLoopGroup(worker)
+          .channelType(EpollServerDomainSocketChannel::class.java);
 
-        builder to {
+        {
           boss.shutdownGracefully()
           worker.shutdownGracefully()
         }

java-plugin/src/main/kotlin/io/github/danielpeach/plugin/ChannelProvider.kt

@@ -46,15 +46,15 @@ class BrokerTest : JUnit5Minutests {
       }
     }
 
-    test("acceptAndServe throws a BrokerServiceException when the broker server is not running") {
+    test("acceptAndServe throws a CancellationException when the broker server is not running") {
       // Client
       brokerClient.start()
 
       // Server
       brokerServer.shutdownAndAwait()
 
       val serviceId = brokerClient.getNextId()
-      expectThrows<BrokerServiceException> {
+      expectThrows<CancellationException> {
         brokerClient.acceptAndServe(serviceId, TestService())
       }
     }

java-plugin/src/test/kotlin/io/github/danielpeach/plugin/BrokerTest.kt

@@ -18,7 +18,7 @@ class ControllerTest : JUnit5Minutests {
 
     after {
       if (!controllerServer.isShutdown) {
-        controllerServer.shutdown()
+        controllerServer.shutdownAndAwait()
       }
     }
 
@@ -62,8 +62,7 @@ class ControllerTest : JUnit5Minutests {
       .start()
       .also { server ->
         controllerService.shutdownHook = {
-          server.shutdownNow()
-          server.awaitTermination()
+          server.shutdownAndAwait()
         }
       }