Connection prematurely closed BEFORE response on half closed connection

When using webclient and calling rest api on another server(using tomcat) the webclient sometimes doesnt acknowledge the connection finish from the server and later try to reuse the already closed connection. The connection is closed by the tomcat server after 60s (default keep alive settings). Most of the times the connection gets closed correctly on the client side but sometimes it just sends [ACK] and no [ACK,FIN] and keeps the connection opened.

Expected Behavior

The client should close the connection after receiving [FIN] from the server.

Actual Behavior

The connection is not closed by the client and later reused resulting in reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response

Full stacktrace:

org.springframework.web.reactive.function.client.WebClientRequestException: Connection prematurely closed BEFORE response; nested exception is reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response
	at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:141)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	|_ checkpoint ⇢ Request to GET http://testload700-metadata-service:8080/api/aggregations/aggregation-ids?entityId=RXBnOjQxNTM3OTIy [DefaultWebClient]
Stack trace:
		at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:141)
		at reactor.core.publisher.MonoErrorSupplied.subscribe(MonoErrorSupplied.java:70)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221)
		at reactor.core.publisher.MonoNext$NextSubscriber.onError(MonoNext.java:93)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onError(MonoFlatMapMany.java:204)
		at reactor.core.publisher.SerializedSubscriber.onError(SerializedSubscriber.java:124)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.whenError(FluxRetryWhen.java:224)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onError(FluxRetryWhen.java:273)
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:413)
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:250)
		at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:491)
		at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:299)
		at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:97)
		at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:189)
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:189)
		at reactor.netty.http.client.HttpClientConnect$HttpObserver.onUncaughtException(HttpClientConnect.java:358)
		at reactor.netty.ReactorNetty$CompositeConnectionObserver.onUncaughtException(ReactorNetty.java:647)
		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onUncaughtException(DefaultPooledConnectionProvider.java:214)
		at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onUncaughtException(DefaultPooledConnectionProvider.java:462)
		at reactor.netty.http.client.HttpClientOperations.onInboundClose(HttpClientOperations.java:290)
		at reactor.netty.channel.ChannelOperationsHandler.channelInactive(ChannelOperationsHandler.java:74)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)
		at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
		at io.netty.handler.codec.http.HttpContentDecoder.channelInactive(HttpContentDecoder.java:235)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelInactive(CombinedChannelDuplexHandler.java:418)
		at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:389)
		at io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:354)
		at io.netty.handler.codec.http.HttpClientCodec$Decoder.channelInactive(HttpClientCodec.java:311)
		at io.netty.channel.CombinedChannelDuplexHandler.channelInactive(CombinedChannelDuplexHandler.java:221)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)
		at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901)
		at io.netty.channel.AbstractChannel$AbstractUnsafe$8.run(AbstractChannel.java:831)
		at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
		at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
		at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:384)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
		at java.base/java.lang.Thread.run(Unknown Source)
Caused by: reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response

Steps to Reproduce

Use webclient to call rest api on a tomcat server. We are sending around 700 requests/s, 300 000 request in total and we get 1 error mentioned before.

There are around 7000 correctly closed connections and single incorrect one so its really uncommon but we can reproduce it in 100% of cases.

This is the webclient with no special configuration:

public class WebClient {
private final ParameterizedTypeReference<Set<Foo>> reference =
      new ParameterizedTypeReference<>() {
      };

public WebClient(
      WebClient.Builder webClientBuilder,
      String serviceUrl
  ) {
    webClient = webClientBuilder.baseUrl(serviceUrl)
        .build();
  }


  private Mono<Set<Foo>> foo(String entityId) {
    UriComponentsBuilder builder = UriComponentsBuilder.fromPath(PATH)
        .query("entityId={entityId}");

    return webClient.get()
        .uri(builder.build()
            .toUriString(), entityId)
        .retrieve()
        .bodyToMono(reference);
  }
}

Possible Solution

Your Environment

Spring Boot Starter Webflux: 2.4.5 Spring Boot Starter Reactor Netty: 2.4.5 Reactor Core: 3.4.5

Java version: openjdk version “11.0.6” 2020-01-14 OpenJDK Runtime Environment AdoptOpenJDK (build 11.0.6+10) OpenJDK 64-Bit Server VM AdoptOpenJDK (build 11.0.6+10, mixed mode)

Running in docker: Linux 27e260f5943f 4.19.0-0.bpo.8-amd64 #1 SMP Debian 4.19.98-1~bpo9+1 (2020-03-09) x86_64 GNU/Linux

Here are the relevant logs: logs_closed_before.csv

And here is the tcp dump: tcp_dump.csv Client is: 172.18.0.18 Server is: 172.18.0.13