reactor-netty: Getting a lot of `PrematureCloseException : Connection prematurely closed BEFORE response`

version: v0.9.7

We are receiving a lot of:

reactor.core.ReactiveException: reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response
    at reactor.core.Exceptions.propagate(Exceptions.java:393)
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:97)
    at reactor.core.publisher.Mono.block(Mono.java:1678)

The client is built like this:

        final ConnectionProvider connectionProvider =
            ConnectionProvider.builder("platform-client-connection-pool-fixed")
                .maxConnections(1_000)
                .pendingAcquireTimeout(Duration.ofMillis(30_000))
                .maxIdleTime(Duration.ofMillis(120))
                .build();

        final TcpClient mainTcpClient = TcpClient.create(connectionProvider)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 15_000)
            .doOnConnected(conn -> conn
                .addHandlerLast(new ReadTimeoutHandler(
                    120_000,
                    TimeUnit.MILLISECONDS))
                .addHandlerLast(new WriteTimeoutHandler(
                    500,
                    TimeUnit.MILLISECONDS))
                .addHandler(new IdleStateHandler(
                    2_000,
                    500, 5000,
                    TimeUnit.MILLISECONDS)));

        final HttpClient httpClient = HttpClient.from(mainTcpClient)
            .keepAlive(true)
            .wiretap(true)
            .compress(true);

        return WebClient.builder()
            .exchangeStrategies(ExchangeStrategies.builder().codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(-1)).build())
            .baseUrl(url)
            .filter(errorHandlingFilter(platformService))
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON)
            .clientConnector(new ReactorClientHttpConnector(httpClient))
            .build();

and the filter is:

    private ExchangeFilterFunction errorHandlingFilter(final PlatformService service) {
        return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
            if (!isErrorStatusCode(clientResponse)) {
                return Mono.just(clientResponse);
            }

            final Optional<org.springframework.http.MediaType> mediaType = clientResponse.headers().contentType();
            boolean jsonResponse = true;

            if (mediaType.isPresent()) {
                final org.springframework.http.MediaType mt = mediaType.get();

                if (!mt.getType().equals("application") && !mt.getSubtype().equals("json")) {
                    jsonResponse = false;
                }
            }

            if (!clientResponse.statusCode().is5xxServerError()) {
                return clientResponse
                    .bodyToMono(String.class)
                    .flatMap(s -> Mono.error(new UpstreamException(s, service)));
            }

            if (!jsonResponse) {
                return clientResponse
                    .bodyToMono(String.class)
                    .flatMap(e -> Mono.error(new UpstreamException(e, service)));
            }

            return clientResponse
                .bodyToMono(RuntimeException.class)
                .flatMap(e -> Mono.error(new UpstreamException(e, service)));
        });
    }

Errors are happening randomly, but all the time.

DEBUG LOG:

3041011 [reactor-http-kqueue-138] DEBUG r.netty.http.client.HttpClient - [id: 0xe27945ea, L:/100.64.0.1:56184 ! R:xxx:443] INACTIVE
3041012 [reactor-http-kqueue-138] DEBUG r.n.r.PooledConnectionProvider - [id: 0xe27945ea, L:/100.64.0.1:56184 ! R:xxx:443] onStateChange(GET{uri=/data-store/portfolios/14?includeSummary=true, connection=PooledConnection{channel=[id: 0xe27945ea, L:/100.64.0.1:56184 ! R:xxx:443]}}, [response_incomplete])
3041013 [reactor-http-kqueue-138] WARN  r.n.http.client.HttpClientConnect - [id: 0xe27945ea, L:/100.64.0.1:56184 ! R:xxx:443] The connection observed an error
reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response
3041013 [reactor-http-kqueue-138] DEBUG r.netty.http.client.HttpClient - [id: 0xe27945ea, L:/100.64.0.1:56184 ! R:xxx:443] UNREGISTERED
3041013 [reactor-http-kqueue-138] DEBUG r.n.r.PooledConnectionProvider - [id: 0x896ef363] Created a new pooled channel, now 456 active connections and 0 inactive connections