Reactor-netty: Connection reset by peer goes in loop

[nidhi7]

Expected behavior

On getting "IOException: Connection reset by peer" again and again, WebClient GET call should exit after 1 or a few retries.

Actual behavior

Reactor netty retries GET calls indefinitely on getting "IOException: Connection reset by peer" again and again. We do not want this to happen as we have our own retry logic added to WebClient call and want that to be executed instead of reactor netty's retries. In our tests, webclient keeps executing the same call and runs out of jvm memory.
If we do the same using spring resttemplate, it works perfectly fine. It throws ResourceAccessException and exits successfully.

import com.github.tomakehurst.wiremock.WireMockServer;
import com.github.tomakehurst.wiremock.client.MappingBuilder;
import com.github.tomakehurst.wiremock.client.WireMock;
import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
import com.github.tomakehurst.wiremock.http.Fault;
import com.github.tomakehurst.wiremock.http.RequestMethod;
import com.github.tomakehurst.wiremock.matching.RequestPatternBuilder;
import io.netty.channel.ChannelOption;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.client.reactive.ReactorResourceFactory;
import org.springframework.util.SocketUtils;
import org.springframework.web.reactive.function.client.*;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.channel.BootstrapHandlers;
import reactor.netty.resources.ConnectionProvider;
import reactor.netty.resources.LoopResources;

import javax.net.ssl.SSLException;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;

public class ConnectionReset {

    private static final int PORT = SocketUtils.findAvailableTcpPort();
    private WireMockServer server;
    private WireMock wireMock;
    private ReactorResourceFactory reactorResourceFactory;
    private String baseUrl;

    @BeforeClass
    public void setup() {
        this.server = new WireMockServer(WireMockConfiguration.wireMockConfig()
                .port(PORT));
        this.server.start();
        this.wireMock = new WireMock(PORT);
        this.wireMock.resetMappings();
        this.baseUrl = "http://localhost:" + PORT;

        this.reactorResourceFactory = new ReactorResourceFactory();
        this.reactorResourceFactory.setUseGlobalResources(false);
        this.reactorResourceFactory.setLoopResourcesSupplier(() -> LoopResources.create("webflux-http-test", 1, true));
        this.reactorResourceFactory.setConnectionProviderSupplier(() -> ConnectionProvider.fixed("test", 1));
        this.reactorResourceFactory.afterPropertiesSet();
    }

    private ClientHttpConnector getClientHttpConnector() {
        return new ReactorClientHttpConnector(this.reactorResourceFactory, httpClient -> httpClient.tcpConfiguration(tcpClient -> {
            try {
                // We are going to load default trust manager
                final SslContext sslContext = SslContextBuilder.forClient()
                        .build();
                tcpClient.secure(t -> t.sslContext(sslContext))
                        .bootstrap(bootstrap -> {
                            BootstrapHandlers.connectionObserver(bootstrap, null);
                            return bootstrap;
                        })
                        .option(ChannelOption.SO_KEEPALIVE, false)
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5_000);
                return tcpClient;
            } catch (final SSLException ex) {
                throw new IllegalStateException("Unable to create SSL context", ex);
            }
        }));
    }

    @AfterClass
    public void tearDown() {
        this.reactorResourceFactory.destroy();
        this.wireMock.shutdown();
        this.server.stop();
    }

    @Test
    public void testReset() {
        final String mockEndpoint = "/my/endpoint";
        final RequestPatternBuilder reqPatternBuilder = RequestPatternBuilder.newRequestPattern(
                RequestMethod.GET, WireMock.urlEqualTo(mockEndpoint));
        final MappingBuilder mappingBuilder = WireMock.get(WireMock.urlEqualTo(reqPatternBuilder.build().getUrl()));

        reqPatternBuilder.withHeader(HttpHeaders.ACCEPT, WireMock.containing(MediaType.APPLICATION_JSON_VALUE))
                .withHeader(HttpHeaders.ACCEPT_CHARSET, WireMock.containing(StandardCharsets.UTF_8.name().toLowerCase()));
        mappingBuilder.willReturn(WireMock.aResponse()
                .withFault(Fault.CONNECTION_RESET_BY_PEER)); // throwing Connection reset 
        this.wireMock.register(mappingBuilder);

        final WebClient webClient = WebClient.builder()
                .baseUrl(this.baseUrl)
                .clientConnector(getClientHttpConnector())
                .build();

        String output = webClient.get()
                .uri(this.baseUrl + mockEndpoint)
                .headers(httpHeaders -> {
                    httpHeaders.add(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE);
                    httpHeaders.add(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
                })
                .retrieve()
                .bodyToMono(String.class)
                .retryWhen(this::untilMaxRetries) // want this to be executed
                .block();
        System.out.println(output);
    }

    // 5 max retries
    public Flux<Long> untilMaxRetries(final Flux<Throwable> throwableFlux) {
        return throwableFlux.zipWith(
                Flux.range(1, 5),
                (ex, index) -> {

                    // I prefer to match more specific exception than IOException.
                    // For eg: PrematureCloseException is a inner class within HttpClientOperations.
                    // It would be good to make them public, so that one can write some behavioral code.
                    if (!(ex instanceof WebClientResponseException) && !(ex instanceof IOException)) {
                        throw Exceptions.propagate(ex);
                    }
                    if (index >= 5) {
                        throw Exceptions.propagate(ex);
                    }
                    return index;
                })
                /*
                 * Delay happens on the default 'parallel' executor.
                 * This is acceptable since the task is just to re-subscribe the post request, which then delegates
                 * work to the executor of the corresponding web client.
                 */
                .flatMap(index -> Mono.delay(Duration.ofMillis(100)));
    }
}

Reactor Netty version

0.8.4.RELEASE

JVM version (e.g. java -version)

java version "1.8.0_131"
Java(TM) SE Runtime Environment (build 1.8.0_131-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.131-b11, mixed mode)

OS version (e.g. uname -a)

16.7.0 Darwin Kernel Version 16.7.0: Sun Oct 28 22:30:19 PDT 2018; root:xnu-3789.73.27~1/RELEASE_X86_64 x86_64

[nidhi7]

Raised this bug under spring https://github.com/spring-projects/spring-framework/issues/22309. Closing this.

[nidhi7]

Reopening this as it is reactor netty bug.

[dilipsundarraj1]

@nidhi7 Did this fix work ? I still face this issue, webClient continuously retrying when the response is "CONNECTION_RESET_BY_PEER".
I am using the below versions of webflux and reactor netty versions.

implementation 'org.springframework:spring-webflux:5.1.8.RELEASE'
    implementation 'io.projectreactor.netty:reactor-netty:0.8.9.RELEASE'

[nidhi7]

Yes it was fixed as far as I remember. I don't use it currently so wouldn't be able to verify quickly. You might want to run the above test and verify.

[violetagg]

@dilipsundarraj1 This was fixed and a test was added so that this issue does not happen again. If you have a reproducible scenario please open a new issue.

[khalDrogo91]

@violetagg
I am also getting this issue with spring boot starter webflux 2.3.1.realese.

Service couldn't respond in time due to which received io exception saying "connection reset by peer". This triggered RETRY.

[violetagg]

@khalDrogo91 Can you provide a reproducible example?

[khalDrogo91]

@violetagg added more details to below link.
https://stackoverflow.com/questions/62800602/io-netty-channel-unix-errorsnativeeceptionioexception-readaddress-failed-conn

[violetagg]

@khalDrogo91 So from that link I get that the problem is not that it goes in a loop, but that the request will be retried. Is that correct?

[violetagg]

@khalDrogo91 You can use this https://projectreactor.io/docs/netty/release/api/reactor/netty/http/client/HttpClient.html#disableRetry-boolean-

[khalDrogo91]

@violetagg it worked like a charm.
Really appreciate your help on it.