Below are the different ways i tried to use web client to send a post, but each time the connection pool is exhausted after sometime. I get the below error.
I'm using spring boot 2.7.3 version
Please help where are the connections not getting released.
The web client initialization code
@PostConstruct
public void getWebClient() {
logger.debug("Initializing webClient..");
ConnectionProvider provider = ConnectionProvider.builder("fixed")
.metrics(true)
.maxConnections(50)
.maxIdleTime(Duration.ofSeconds(20))
.maxLifeTime(Duration.ofSeconds(60))
.pendingAcquireTimeout(Duration.ofSeconds(30))
.evictInBackground(Duration.ofSeconds(30)).build();
HttpClient httpClient = HttpClient.create(provider)
.wiretap(this.getClass().getCanonicalName(), LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
.responseTimeout(Duration.ofSeconds(10));
this.webClient = WebClient
.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient))
.defaultHeader(HttpHeaders.CONTENT_TYPE, org.springframework.http.MediaType.APPLICATION_JSON_VALUE)
.build();
}
Code 1 :
public Mono<JsonNode> sendPostR(String url, JsonNode request) {
String uuid = UUID.randomUUID().toString();
Instant start = Instant.now();
logger.debug("Sending post request {} {} {}",url,request,uuid);
return webClient.post().uri(url)
.body(Mono.just(request),Map.class)
.retrieve()
.toEntity(String.class)
.map(entity -> {
Instant end = Instant.now();
logger.debug("Time taken for request success{} {} ",Duration.between(start,end).toMillis(),uuid);
if(entity.getStatusCode() != HttpStatus.OK) {
logger.error("Error: API returned with in-valid status code {} {}",entity.getStatusCode(),entity.getHeaders());
Mono.error(new RuntimeException("Error in sending post " + url));
}
String responseBody = entity.getBody();
try {
return objectMapper.readValue(responseBody, JsonNode.class);
} catch (JsonProcessingException e) {
logger.error("Error in parsing response body",e);
Mono.error(new RuntimeException("Error in parsing response body " + url));
}
return Utils.getNewObjectNode();
}).switchIfEmpty(Mono.defer(() -> {
logger.debug("Request completed with no body response {}",uuid);
return Mono.just(Utils.getNewObjectNode());
})).log();
}
Code 2:
public Mono<JsonNode> sendPostR(String url, Map<String,Object> params) {
String uuid = UUID.randomUUID().toString();
Instant start = Instant.now();
logger.debug("Sending post request {} {} {}",url,params,uuid);
return webClient.post().uri(url)
.body(Mono.just(params),Map.class)
.exchangeToMono(response -> {
logger.debug("response is {}",response.statusCode());
Instant end = Instant.now();
logger.debug("Time taken for request success{} {} ",Duration.between(start,end).toMillis(),uuid);
if(response.statusCode().equals(HttpStatus.OK)) {
return response.bodyToMono(JsonNode.class);
}
response.releaseBody();
logger.debug("Error in hitting url {}",url);
return Mono.error(new Exception("Error in sending post " + url));
}).onErrorResume(WebClientResponseException.class, ex -> {
Instant end = Instant.now();
logger.debug("Time taken for request error {} {} ",Duration.between(start,end).toMillis(),uuid);
logger.debug("Error in hitting url .... ",ex);
return Mono.error(ex);
}).doOnError(error -> {
logger.error("Error in sending post request ",error);
}).switchIfEmpty(Mono.defer(() -> {
logger.debug("Request completed with no body response {}",uuid);
return Mono.just(Utils.getNewObjectNode());
}))
.log();
Below is the code, which i use if i do not need the response. I use for fire and forget scenarios.
public Mono<ResponseEntity<Void>> raiseEvent(String url, JsonNode request) {
String uuid = UUID.randomUUID().toString();
logger.debug("Sending post request {} {} {}",url,request,uuid);
return webClient.post().uri(url)
.body(Mono.just(request),JsonNode.class)
.retrieve()
.toBodilessEntity() ;
}
Error which i get when pool is exhausted.
2022-09-22 15:07:52,701 ERROR ? [parallel-1] Operator called default onErrorDropped
reactor.core.Exceptions$ErrorCallbackNotImplemented: org.springframework.web.reactive.function.client.WebClientRequestException: Pool#acquire(Duration) has been pending for more than t
he configured timeout of 30000ms; nested exception is reactor.netty.internal.shaded.reactor.pool.PoolAcquireTimeoutException: Pool#acquire(Duration) has been pending for more than the
configured timeout of 30000ms
Caused by: org.springframework.web.reactive.function.client.WebClientRequestException: Pool#acquire(Duration) has been pending for more than the configured timeout of 30000ms; nested e
xception is reactor.netty.internal.shaded.reactor.pool.PoolAcquireTimeoutException: Pool#acquire(Duration) has been pending for more than the configured timeout of 30000ms
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 POST http://api.call/track [DefaultWebClient]
Original 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:55)
at reactor.core.publisher.Mono.subscribe(Mono.java:4397)
at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
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:225)
at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onError(FluxRetryWhen.java:274)
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:415)
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:251)
at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:537)
at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:343)
at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100)
at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:190)
at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:201)
at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect$ClientTransportSubscriber.onError(HttpClientConnect.java:308)
at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:201)
at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onError(DefaultPooledConnectionProvider.java:158)
at reactor.netty.internal.shaded.reactor.pool.AbstractPool$Borrower.run(AbstractPool.java:413)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: reactor.netty.internal.shaded.reactor.pool.PoolAcquireTimeoutException: Pool#acquire(Duration) has been pending for more than the configured timeout of 30000ms
at reactor.netty.internal.shaded.reactor.pool.AbstractPool$Borrower.run(AbstractPool.java:413)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)