Tapir Netty Graceful Shutdown Bug

README.md

Minimal reproduction of the issue of tapir-netty server not shutting down gracefully while there are in-flight requests.

Result:

[Server] started.
[Client] Sending request.
[Server] received the request.
[Client] Stopping server.
sttp.client3.SttpClientException$ConnectException: Exception when sending request: GET http://localhost:8080/hello
	at sttp.client3.SttpClientExceptionExtensions.defaultExceptionToSttpClientException(SttpClientExceptionExtensions.scala:13)
	at sttp.client3.SttpClientExceptionExtensions.defaultExceptionToSttpClientException$(SttpClientExceptionExtensions.scala:11)
	at sttp.client3.SttpClientException$.defaultExceptionToSttpClientException(SttpClientException.scala:24)
	at sttp.client3.HttpClientAsyncBackend.$anonfun$adjustExceptions$1(HttpClientAsyncBackend.scala:147)
	at sttp.client3.SttpClientException$$anonfun$adjustExceptions$1.applyOrElse(SttpClientException.scala:35)
	at sttp.client3.SttpClientException$$anonfun$adjustExceptions$1.applyOrElse(SttpClientException.scala:34)
	at scala.concurrent.impl.Promise$Transformation.run$$$capture(Promise.scala:490)
	at scala.concurrent.impl.Promise$Transformation.run(Promise.scala)
	at java.base/java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1426)
	at java.base/java.util.concurrent.ForkJoinTask.doExec$$$capture(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
Caused by: java.net.ConnectException: Connection refused
	at java.base/sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
	at java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:777)
	at java.net.http/jdk.internal.net.http.PlainHttpConnection$ConnectEvent.handle(PlainHttpConnection.java:128)
	at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.handleEvent(HttpClientImpl.java:957)
	at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.lambda$run$3(HttpClientImpl.java:912)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
	at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.run(HttpClientImpl.java:912)
[Client] Stopped server.
[Server] successfully processed the request.
[Application] shutdown signal received.

Expectation:

[Server] started.
[Client] Sending request.
[Server] received the request.
[Client] Stopping server.
[Server] successfully processed the request.
[Client] Response received: Response(Right(Hello World!),200,,List(content-length: 12, content-type: text/plain; charset=UTF-8),List(),RequestMetadata(GET,http://localhost:8080/hello,Vector()))
[Client] Stopped server.
[Application] shutdown signal received.

build.sbt

ThisBuild / version := "0.1.0-SNAPSHOT"
ThisBuild / scalaVersion := "2.13.11"

lazy val root = (project in file("."))
  .settings(
    libraryDependencies ++= Seq(
      "com.softwaremill.sttp.tapir" %% "tapir-netty-server" % "1.5.5",
      "com.softwaremill.sttp.tapir" %% "tapir-sttp-client" % "1.5.5"
    ),
    name := "scala-tapir-netty-shutdown"
  )

logback.xml

<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%level %logger{15} - %message%n%xException{10}</pattern>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="STDOUT"/>
    </root>
</configuration>

Main.scala

import sttp.client3._
import sttp.model.Uri
import sttp.tapir._
import sttp.tapir.server.netty.{NettyFutureServer, NettyFutureServerOptions, NettyOptions}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.{Await, Future, blocking}
import scala.util.{Failure, Success}

object Main {

  private val helloWorldEndpoint = endpoint
    .get
    .in("hello")
    .out(stringBody)
    .serverLogicSuccess(_ =>
      Future {
        println("[Server] received the request.")
        blocking {
          Thread.sleep(3000)
          println("[Server] successfully processed the request.")
        }
        s"Hello World!"
      }
    )

  def main(args: Array[String]): Unit = {
    implicit val backend = HttpClientFutureBackend()

    // shutdown hook to print if a signal is received
    Runtime.getRuntime.addShutdownHook(new Thread((() => {
      println("[Application] shutdown signal received.")
    }):Runnable))

    // start server
    val serverOptions = NettyFutureServerOptions.default
      .nettyOptions(
        NettyOptions.default
          .copy(shutdownEventLoopGroupOnClose = true) // no effect as it is enabled by default
      )
    val bindingF = NettyFutureServer().port(8080).options(serverOptions).addEndpoint(helloWorldEndpoint).start()
    val binding = Await.result(bindingF, 10.seconds)
    println("[Server] started.")


    // call my endpoint and then kill me
    println(s"[Client] Sending request.")
    emptyRequest
      .get(Uri.parse("http://localhost:8080/hello").getOrElse(???))
      .send(backend)
      .onComplete {
        case Success(r) => println(s"[Client] Response received: $r")
        case Failure(exception) => exception.printStackTrace()
      }
    // wait until the service receives the request
    Thread.sleep(1000L)

    // kill myself
    println(s"[Client] Stopping server.")
    Await.result(binding.stop(), 10.seconds)

    println(s"[Client] Stopped server.")
  }

}

Posted this reproduction scenario to Tapir project

softwaremill/tapir#3000