upgrade to Ktor 3

Author: ianbotsf

gradle/libs.versions.toml

@@ -92,7 +92,7 @@ ktor-http-cio = { module = "io.ktor:ktor-http-cio", version.ref = "ktor-version"
 ktor-utils = { module = "io.ktor:ktor-utils", version.ref = "ktor-version" }
 ktor-io = { module = "io.ktor:ktor-io", version.ref = "ktor-version" }
 ktor-server-netty = { module = "io.ktor:ktor-server-netty", version.ref = "ktor-version" }
-ktor-server-jetty = { module = "io.ktor:ktor-server-jetty", version.ref = "ktor-version" }
+ktor-server-jetty-jakarta = { module = "io.ktor:ktor-server-jetty-jakarta", version.ref = "ktor-version" }
 ktor-server-cio = { module = "io.ktor:ktor-server-cio", version.ref = "ktor-version" }
 ktor-network-tls-certificates = { module = "io.ktor:ktor-network-tls-certificates", version.ref = "ktor-version" }
 

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/ConnectionIdleMonitor.kt

@@ -4,7 +4,6 @@
  */
 package aws.smithy.kotlin.runtime.http.engine.okhttp
 
-import aws.smithy.kotlin.runtime.telemetry.logging.Logger
 import aws.smithy.kotlin.runtime.telemetry.logging.logger
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
@@ -63,19 +62,20 @@ internal class ConnectionIdleMonitor(val pollInterval: Duration) : ConnectionLis
         val connId = System.identityHashCode(connection)
         val context = call.callContext()
         val scope = CoroutineScope(context)
-        val logger = context.logger<ConnectionIdleMonitor>()
         val monitor = scope.launch(CoroutineName("okhttp-conn-monitor-for-$connId")) {
-            doMonitor(connection, logger)
+            doMonitor(connection)
         }
-        logger.trace { "Launched coroutine $monitor to monitor $connection" }
+        context.logger<ConnectionIdleMonitor>().trace { "Launched coroutine $monitor to monitor $connection" }
 
         // Non-locking map access is okay here because this code will only execute synchronously as part of a
         // `connectionReleased` event and will be complete before any future `connectionAcquired` event could fire for
         // the same connection.
         monitors[connection] = monitor
     }
 
-    private suspend fun doMonitor(conn: Connection, logger: Logger) {
+    private suspend fun doMonitor(conn: Connection) {
+        val logger = coroutineContext.logger<ConnectionIdleMonitor>()
+
         val socket = conn.socket()
         val source = try {
             socket.source()
@@ -93,16 +93,20 @@ internal class ConnectionIdleMonitor(val pollInterval: Duration) : ConnectionLis
 
             while (coroutineContext.isActive) {
                 try {
+                    logger.trace { "Polling socket for $conn" }
                     source.readByte() // Blocking read; will take up to READ_TIMEOUT_MS to complete
                 } catch (_: SocketTimeoutException) {
+                    logger.trace { "Socket still alive for $conn" }
                     // Socket is still alive
                 } catch (_: EOFException) {
-                    logger.trace { "Socket for $conn was closed remotely" }
+                    logger.trace { "Socket closed remotely for $conn" }
                     socket.closeQuietly()
                     resetTimeout = false
                     return
                 }
             }
+
+            logger.trace { "Monitoring coroutine has been cancelled. Ending polling loop." }
         } catch (e: Throwable) {
             logger.warn(e) { "Failed to poll $conn. Ending polling loop. Connection may be unstable now." }
         } finally {

runtime/protocol/http-client-engines/test-suite/build.gradle.kts

@@ -24,7 +24,7 @@ kotlin {
 
         jvmMain {
             dependencies {
-                implementation(libs.ktor.server.jetty)
+                implementation(libs.ktor.server.jetty.jakarta)
                 implementation(libs.ktor.network.tls.certificates)
 
                 implementation(project(":runtime:protocol:http-client-engines:http-client-engine-default"))
@@ -52,6 +52,8 @@ kotlin {
                 implementation("org.bouncycastle:bcpkix-jdk18on:1.78") // https://github.com/docker-java/docker-java/pull/2326
 
                 implementation(libs.docker.transport.zerodep)
+
+                implementation(project(":runtime:observability:telemetry-defaults"))
             }
         }
 

runtime/protocol/http-client-engines/test-suite/jvm/src/aws/smithy/kotlin/runtime/http/test/suite/Connections.kt

@@ -5,33 +5,16 @@
 package aws.smithy.kotlin.runtime.http.test.suite
 
 import io.ktor.server.application.Application
-import io.ktor.server.application.call
-import io.ktor.server.jetty.JettyApplicationCall
 import io.ktor.server.response.respondText
-import io.ktor.server.routing.RoutingApplicationCall
 import io.ktor.server.routing.post
 import io.ktor.server.routing.route
 import io.ktor.server.routing.routing
-import io.ktor.util.InternalAPI
-import io.ktor.utils.io.close
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
-import kotlin.time.Duration.Companion.seconds
 
 internal fun Application.connectionTests() {
     routing {
         route("connectionDrop") {
-            @OptIn(InternalAPI::class)
             post {
-                val routingCall = call as RoutingApplicationCall
-                val jettyCall = routingCall.engineCall as JettyApplicationCall
-
-                launch {
-                    delay(4.seconds) // Close the connection ~4 seconds after the call ends
-                    jettyCall.response.responseChannel().close()
-                }
-
-                jettyCall.respondText("Bar")
+                call.respondText("Bar")
             }
         }
     }

runtime/protocol/http-client-engines/test-suite/jvm/src/aws/smithy/kotlin/runtime/http/test/util/TestServers.kt

@@ -13,10 +13,13 @@ import aws.smithy.kotlin.runtime.http.test.suite.uploadTests
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
 import io.ktor.server.jetty.*
+import io.ktor.server.jetty.jakarta.Jetty
+import io.ktor.server.jetty.jakarta.JettyApplicationEngineBase
 import redirectTests
 import java.io.Closeable
 import java.nio.file.Paths
 import java.util.concurrent.TimeUnit
+import kotlin.time.Duration.Companion.seconds
 
 private data class TestServer(
     val port: Int,
@@ -94,7 +97,7 @@ private fun tlsServer(instance: TestServer, sslConfig: SslConfig): EmbeddedServe
     val rootConfig = serverConfig {
         module(instance.initializer)
     }
-    val engineConfig: ApplicationEngine.Configuration.() -> Unit = {
+    val engineConfig: JettyApplicationEngineBase.Configuration.() -> Unit = {
         when (instance.type) {
             ConnectorType.HTTP -> connector { port = instance.port }
 
@@ -109,6 +112,8 @@ private fun tlsServer(instance: TestServer, sslConfig: SslConfig): EmbeddedServe
                 enabledProtocols = instance.protocolName?.let(::listOf)
             }
         }
+
+        idleTimeout = 3.seconds // Required for ConnectionTest.testShortLivedConnections
     }
 
     return try {

runtime/protocol/http-client-engines/test-suite/jvm/test/aws/smithy/kotlin/runtime/http/test/ConnectionTest.kt

@@ -109,7 +109,7 @@ class ConnectionTest : AbstractEngineTest() {
             val initialResp = initialCall.response.body.toByteStream()?.decodeToString()
             assertEquals("Bar", initialResp)
 
-            delay(5.seconds) // Shorter than the client-side timeout
+            delay(5.seconds) // Longer than the service side timeout, shorter than the client-side timeout
 
             val subsequentReq = HttpRequest {
                 testSetup()