Getting RetryExhaustedException while upserting bulk records in couchbase

[SuhasT]

Hello,

We are upgrading couchbase SDK from 2 to 3.4.2 and while upgrading in one component getting error to bulk upsert.

Error - reactor.core.Exceptions$RetryExhaustedException: Retries exhausted: 3/3

Sample Code in SDK3 -

Flux.fromIterable(documents)
.flatMap(jsonDocument → {
return collection.reactive().upsert(jsonDocument.getString(“id”), jsonDocument)
.retryWhen(Retry.max(MAX_RETRIES)
.fixedDelay(MAX_RETRIES, Duration.ofMillis(BASE_DELAY))
.filter(t → t instanceof CouchbaseException)
)
.doOnError(exception → System.out.println("\n Error : "+ exception))
.onErrorResume(error → Mono.empty());
})
.count()
.single()
.block()
.intValue();

Can someone please help to fix this issue ?

[ingenthr]

There isn’t a lot of context, but it looks like you are allowing for 3 retries and that isn’t enough based on your workload. I’d recommend increasing your duration and/or number of retries.

I’d also note that exponential (see the “backoff” on reactor) may be better than fixed for this kind of situation.

[SuhasT]

Hi @ingenthr ,

Yes, we are having Max retries count configured as 3. we even increased this to 5 and still getting same error.

As you suggested, we tried with backoff on reactor as well, as still seeing same issue. Here is the sample code snippet -

	Integer insertCount = Flux.fromIterable(documents).publishOn(Schedulers.boundedElastic())
			.flatMap(jsonDocument -> {
		return collection.reactive().upsert(jsonDocument.getString("id"), jsonDocument)
				.retryWhen(Retry.backoff(10, Duration.ofMillis(10))
				             		.filter(t -> t instanceof CouchbaseException)
				)
				.doOnError(exception -> System.out.println("\n Error : "+ exception))
				.onErrorResume(error -> Mono.empty());
	})
	.count()
	.single()
	.block()
	.intValue(); 

Thanks