Reactor-netty: Connection not released in ConnectionPool when PrematureCloseException occurs

[rreynaud]

When HttpClient made requests and PrematureCloseException occurs, connections are not released in ConnectionPool and stayed always in Active state. I suspect it may occur with other exceptions but I have not tried to reproduce

Expected Behavior

We expect the connection are released or removed from the ConnectionPool and do not take any slots.

Actual Behavior

ConnectionPool is growing until reaching maximum capacity with active connections which leads to a saturated pool and future acquisition problems.

Steps to Reproduce

Provide an HttpServer with an Https scheme and use HttpClient to make request in Http scheme and look at the ConnectionPoolProvider traces.

ServerSide code:

public static void main(final String[] args) throws InterruptedException, CertificateException {
    final SelfSignedCertificate cert = new SelfSignedCertificate();
    final SslContextBuilder sslContextBuilder = SslContextBuilder.forServer(cert.key(), cert.cert());
    final CountDownLatch latch = new CountDownLatch(1);
    HttpServer.create()
              .secure(s -> s.sslContext(sslContextBuilder))
              .port(45454)
              .route(r -> r.route(p -> true, (req, resp) -> resp.sendHeaders()))
              .bindNow();

    latch.await(5, TimeUnit.MINUTES);
}

ClientSide code:

public static void main(final String[] args) throws SSLException, InterruptedException {
    final SslContext insecured = SslContextBuilder.forClient()
                                                  .trustManager(InsecureTrustManagerFactory.INSTANCE)
                                                  .build();
    HttpClient client = HttpClient.create()
                                  .secure(s -> s.sslContext(insecured));
    final CountDownLatch latch = new CountDownLatch(1);
    Flux.range(1, 510)
        .flatMap(i -> client.get()
                            .uri("http://localhost:45454/ping")
                            .response()
                            .onErrorResume(e -> Mono.empty())
                            .log(), 2)
        .then()
        .doFinally(s -> latch.countDown())
        .subscribe();

    latch.await(5, TimeUnit.MINUTES);
}

Your Environment

• Reactor version(s) used: 0.9.1.RELEASE
• Other relevant libraries versions (eg. netty, ...): ReactorCore 3.3.0.RELEASE
• JVM version (javar -version): JDK 11.0.2
• OS and version (eg uname -a): Kubuntu 18.04

[jim2paterson]

This looks very similar to an error I would like to report but have not been able to recreate in a simple example. @rreynaud , I ran your code and got the error. Of course, removing the SSL / secure parts it works. Then I modified your client to resemble what I'm trying to do. I don't mean to complicate this issue, but I want to share some of my test results. In the client side code below, it fails with the runOn() and works when it is commented out. I would very much like to know why.

public static void main(final String[] args) throws SSLException, InterruptedException {
        final TcpClient tcpClient = TcpClient
                .create(ConnectionProvider.fixed("thepool", 10, 1))
//                .runOn(new NioEventLoopGroup(5))
                .wiretap(true);
        final SslContext insecured = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE)
                .build();
        HttpClient client = HttpClient.from(tcpClient)
                .secure(s -> s.sslContext(insecured));
        final CountDownLatch latch = new CountDownLatch(1);
        Flux.range(1, 510)
                .flatMap(i -> client.get()
                        .uri("http://localhost:45454/ping")
                        .response()
                        .onErrorResume(e -> Mono.empty())
                        .log(), 2)
                .then()
                .doFinally(s -> latch.countDown())
                .subscribe();

        latch.await(5, TimeUnit.MINUTES);
    }

[violetagg]

@rreynaud PR #909

@jim2paterson
I do not quite understand the problem, but I can say that you should create daemon threads.
Use some of the other NioEventLoopGroup constructors OR you can use the API that Reactor Netty provides i.e.:

.runOn(LoopResources.create("test", 5, true))

Another idea:

NioEventLoopGroup group = new NioEventLoopGroup(5);
...
.runOn(group)
...
.doFinally(s -> {
    group.shutdownGracefully();
    latch.countDown();
})

In this case you even do not need the latch

[rreynaud]

@violetagg Thanks