How to handle "io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer" in WebClient Springboot

I had 579,000 pending messages in my ActiveMQ, and I needed to publish these messages to a REST API. While doing so I was hitting the REST API at the rate of 3000 hits/second asynchronously. After some time I started getting the below mentioned exception continously for every request:-

2021:06:10 05:25:40.002 [DefaultMessageListenerContainer-58047] [INFO] [com.main.consumer.AmazonMQConsumer] - Recieved TOPIC MESSAGE: DUMMY_MSG
2021:06:10 05:25:40.004 [reactor-http-epoll-6] [WARN] [reactor.netty.http.client.HttpClientConnect] - [id: 0DUMM895fb, L:/DUMMY_IP:DUMMY_PORT - R:DUMMY-URL.net/DUMMY_IP':DUMMY_PORT'] The connection observed an error
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
2021:06:10 05:25:40.173 [reactor-http-epoll-7] [WARN] [reactor.netty.http.client.HttpClientConnect] - [id: 0DUMMf5600, L:/DUMMY_IP:DUMMY_PORT - R:DUMMY-URL.net/DUMMY_IP':DUMMY_PORT'] The connection observed an error
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
2021:06:10 05:25:40.365 [reactor-http-epoll-9] [WARN] [reactor.netty.http.client.HttpClientConnect] - [id: 0DUMM9223f, L:/DUMMY_IP:DUMMY_PORT - R:DUMMY-URL.net/DUMMY_IP':DUMMY_PORT'] The connection observed an error
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
2021:06:10 05:25:41.966 [reactor-http-epoll-8] [WARN] [reactor.netty.http.client.HttpClientConnect] - [id: 0DUMMb5cf1, L:/DUMMY_IP:DUMMY_PORT - R:DUMMY-URL.net/DUMMY_IP':DUMMY_PORT'] The connection observed an error
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
2021:06:10 05:25:43.419 [reactor-http-epoll-8] [WARN] [reactor.netty.http.client.HttpClientConnect] - [id: 0DUMM7fa0a, L:/DUMMY_IP:DUMMY_PORT - R:DUMMY-URL.net/DUMMY_IP':DUMMY_PORT'] The connection observed an error
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
2021:06:10 05:25:43.419 [reactor-http-epoll-8] [WARN] [io.netty.channel.AbstractChannelHandlerContext] - An exception 'reactor.core.Exceptions$ErrorCallbackNotImplemented: reactor.core.Exceptions$RetryExhaustedException: Retries exhausted: 3/3' [enable DEBUG level for full stacktrace] was thrown by a user handler's exceptionCaught() method while handling the following exception:
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    |_ checkpoint â‡¢ Request to POST https://DUMMY-URL.net/api/v1/evt?p=.....so on
    Stack trace:

Asychronous Rest API call code:-

@Service
public class DummyConnectionService {
    @Value("${web.client.retry.max-attempts}")
    private String maxAttempts;
    
    @Value("${web.client.retry.min-backoff}")
    private String minBackoff;
    
    @Value("${Dummy.base-url}")
    private String DummyBaseUrl;
    
    @Value("${Dummy.api-path}")
    private String DummyApiPath;
    
    @Autowired
    WebClient.Builder webClientBuilder;
    
    public Mono<String> sendDataToDummy(String query, final String eventData) throws Exception {
        Mono<String> response = webClientBuilder.baseUrl(DummyBaseUrl)
                .build()
                .post()
                .uri(uriBuilder -> uriBuilder
                        .path(DummyApiPath)
                        .query(query)
                        .build(eventData))
                .retrieve()
                .bodyToMono(String.class)
                .retryWhen(Retry.backoff(Long.parseLong(maxAttempts), Duration.ofSeconds(Long.parseLong(minBackoff)))); //maxAttempts = 3, minBackoff = 5
        
        return response;
    }
}

Calling code for sendDataToDummy():-

Mono<String> response = DummyConnectionService.sendDataToDummy(postData.toString(), json);
            
response.subscribe(res -> {
    if(res.contains(STATUS_OK)) {
        log.info(res);
    } else {
        log.error(res);
    }
});

Kindly help me by letting me know why such exception is coming and how to handle it gracefully without halting program flow.

Recommended topics

Hot tags