java.util.concurrent.TimeoutException: Connection lease request time out #2492
Comments
spring-projects-issues
added
the
status: waiting-for-triage
|
Can you provide a compilable/runnable project? With what call are you saving the objects? |
sothawo
added
status: waiting-for-feedback
|
Hello @sothawo , Thank you for taking a look at this. |
spring-projects-issues
added
status: feedback-provided
|
Ok, that's a nice and interesting problem. First let me explain how the public <T> Mono<IndexResponse> index(IndexRequest<T> request) {
Assert.notNull(request, "request must not be null");
return Mono.fromFuture(transport.performRequestAsync(request, IndexRequest._ENDPOINT, transportOptions));
}Whatever is done in this call is outside of Spring Data Elasticsearch . These async methods return a When looking at your code: you take every single message, cobnvert that to an entity object and send that off to Elasticsearch, then returning the saved entity in a flux. First, this is extremely inefficient, as you produce a separate network call for every single object. To reproduce this, I used a sample application using a local Elasticsearch provided by docker, I did not bother to set up a Kafka. My test entity is this: @Document(indexName = "foo")
public class Foo {
@Id
@Nullable
private String id;
@Nullable
private String text;
public static Foo of(int id) {
var foo = new Foo();
foo.setId("" + id);
foo.setText("text-" + id);
return foo;
}
@Nullable
public String getId() {
return id;
}
public void setId(@Nullable String id) {
this.id = id;
}
@Nullable
public String getText() {
return text;
}
public void setText(@Nullable String text) {
this.text = text;
}
}The repository is the same as in your case. A simple test method to reproduce this error: public Flux<Foo> test1() {
return Flux.range(1, 1_000_000)
.map(Foo::of)
.flatMapSequential(repository::save);
}So, what to do? The first step is to group the incoming request into batches of lets say 1000 entities, waiting at most one second: public Flux<Foo> test1() {
return Flux.range(1, 1_000_000)
.map(Foo::of)
.bufferTimeout(1000, Duration.ofSeconds(1))
.flatMapSequential(repository::saveAll);
}Nice try, but same error, but only later. Pushing in a million objects still leads to 1000 calls which is too much. We need to add backpressure handling and this requires us to write quite some more code (we ignore returning a flux of entites here): public Flux<Foo> test2() {
Flux.range(1, 1_000_000)
.map(Foo::of)
.bufferTimeout(1000, Duration.ofSeconds(1))
.subscribe(new Subscriber<>() {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(1);
}
@Override
public void onNext(List<Foo> foos) {
repository.saveAll(foos)
.doOnComplete(() -> {
subscription.request(1);
})
.subscribe();
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
subscription.cancel();
throw new RuntimeException("oops, something went wrong", throwable);
}
@Override
public void onComplete() {
}
});
return Flux.empty();
}We are adding backpressure handling here: We still batch the incoming objects into groups of 1000, but only request the first of that. When we get it, we store it in Elasticsearch , when that's done, we request the next batch and so on. As a final step we provide the saved entities as a Flux to the caller, this requires a slightly modified handling at the end of the upstream flux: public Flux<Foo> test3() {
Sinks.Many<Foo> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux.range(1, 1_000_000)
.map(Foo::of)
.bufferTimeout(1000, Duration.ofSeconds(1))
.subscribe(new Subscriber<>() {
private Subscription subscription;
volatile boolean upstreamComplete = false;
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(1);
}
@Override
public void onNext(List<Foo> foos) {
repository.saveAll(foos)
.map(sink::tryEmitNext)
.doOnComplete(() -> {
if (!upstreamComplete) {
subscription.request(1);
} else {
sink.tryEmitComplete();
}
})
.subscribe();
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
subscription.cancel();
sink.tryEmitError(throwable);
}
@Override
public void onComplete() {
upstreamComplete = true;
}
});
return sink.asFlux();
}I could insert 1 million docs into my local dockered Elasticsearch with that in 2.5 minutes (10 year old iMac) with no errors. So no error in Spring Data Elasticsearch, but need of special reactive code. |
sothawo
added
status: waiting-for-feedback
|
Thank you @sothawo for the nice writing. This worries me a little bit unfortunately. I was hoping I did something wrong configuring the repository or something like that. I did another test, same code: Just replaced the reactive elastic repository with 1) a reactive cassandra repository 2) a reactive mongodb repository. Exact same code, just interchanging the reactive repository. For both cassandra and mongo, I could insert the millions of entities with little to no cpu/memory footprint within no time. I believe saving a flux of "something" inside elasticseach should be a common use case. (I searched for tutorials online, and it seems a basic webflux app to save a flux also uses the same approach. does it mean many are encountering the same issue?) Is there a better way to do this without all the boiler plate code? Can I change my Configuration class (maybe changing the template, the operation, the restclient, else) in order to resolve this issue? Our use case is quite simple, just saving a flux of "something" "somewhere". This somewhere being cassandra, mongo or other reactive databases. It seems we have to add this custom code on the save part just for elasticsearch. So I was wondering if I could get your help pointing out if there are other alternatives. |
spring-projects-issues
added
status: feedback-provided
|
Update: I tried again this time with myElasticRepository.saveAll(theFlux) (the flux being the flux coming from Kafka). And again, unfortunately, same issue, while the mongo or elastic repository would save everything fine. I can hardly believe I am the first to save a flux using spring data Elasticsearch reactive. What am I doing wrong please? Many thanks |
|
Comparing with Mongo or Cassandra makes no sense, Mongo has a client that communicates with the server using a native protocol, whereas Spring Data Elasticsearch needs to use the transport provided by Elasticsearch. So as explained before, by sending every item as a separate request - like you did first - results in a HTTP request sent to the server, and that is not handled by the transport provided by Elasticsearch. As for the |
|
Many thanks @sothawo |
Hello team,
I wanted to reach out reporting an issue 100% reproducible with reactive spring data elasticsearch please.
The setup is simple:
The business logic is straightforward, using Reactor-Kafka; Spring Cloud Stream Kafka Reactive, or Spring Webflux, I will receive a flux of MyPojo. My goal is just to save them inside Elastic.
The rate is quite fast, therefore, I am using the reactive repository.
When using above setup, I will get 100%:
The thing is:
I tried ``` .withConnectTimeout() .withSocketTimeout()`` set to some crazy value, still getting this error.
I tried switching to non reactive, and it works.
Finally, I will see this stack trace after processing many messages. Yet, I will see only 1 Hit in Elasticsearch.
May I ask what is the root cause of this issue please?
Thank you
The text was updated successfully, but these errors were encountered: