Jetty HTTP client issue testcase

JettyClientTest.kt

package org.jetbrains.ktor.client

import org.eclipse.jetty.http2.client.*
import org.eclipse.jetty.http2.client.http.*
import org.eclipse.jetty.http2.server.*
import org.eclipse.jetty.server.*
import org.eclipse.jetty.servlet.*
import org.eclipse.jetty.util.thread.*
import org.junit.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
import javax.servlet.http.*
import kotlin.test.*

class JettyClientTest {
    private val transport = HTTP2Client()
    private val client = org.eclipse.jetty.client.HttpClient(HttpClientTransportOverHTTP2(transport), null).apply {
        isConnectBlocking = false
        isStrictEventOrdering = true // tried to switch these settings but no luck
    }

    private var server: Server? = null

    @After
    fun tearDown() {
        server?.stop()
        client.stop()
        transport.stop()
    }

    @Test
    fun smokeTest() {
        server = createServer("/", 9096) { request, response ->
            response.status = HttpServletResponse.SC_OK
            response.contentType = "text/plain;charset=utf-8"
            response.writer.apply {
                for (i in 1..10) {
                    appendln("message $i")
                    flush()
                }
            }
        }

        transport.start()
        client.start()

        val latch = CountDownLatch(1)
        val eventCounter = AtomicInteger()
        val request = client.newRequest("http://localhost:9096/")

        request.onResponseContentAsync { response, buffer, callback ->
            eventCounter.incrementAndGet()
            // don't call callback so we have to receive it only once
        }
        request.send {
            latch.countDown() // we never reach here
        }

        latch.await(10, TimeUnit.SECONDS) // not exact robust for real test, but good enough for example
        assertEquals(1, eventCounter.get()) // here we fail with actual value 11
    }

    companion object {
        val serverExecutor = QueuedThreadPool().apply {
            name = "server"
            start()
        }

        @JvmStatic
        @AfterClass
        fun shutdown() {
            serverExecutor.stop()
        }

        fun createServer(path: String, port: Int, handler: (HttpServletRequest, HttpServletResponse) -> Unit): Server {
            val server = Server(serverExecutor)
            val connector = ServerConnector(server, 1, 1, HTTP2ServerConnectionFactory(HttpConfiguration()))
            connector.port = port
            server.addConnector(connector)

            val context = ServletContextHandler(server, "/", true, false)
            context.addServlet(ServletHolder(object : DefaultServlet() {
                override fun doGet(request: HttpServletRequest, response: HttpServletResponse) {
                    handler(request, response)
                }
            }), path)

            server.start()

            return server
        }
    }
}