Fixed issue with onErrorDropped being called when using concatWith in…

… QuorumReader (Azure#24180)
srnagar · Sep 17, 2021 · 07ab715 · 07ab715
1 parent cae878b
commit 07ab715
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/...cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/QuorumReader.java b/...cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/QuorumReader.java
@@ -242,7 +242,9 @@ public Mono<StoreResponse> readStrongAsync(
                     });
             }).repeat(maxNumberOfReadQuorumRetries)
                    .takeUntil(dummy -> !shouldRetryOnSecondary.v)
-                .concatWith(Flux.defer(() -> {
+                   //   In case there is an empty response from above flatMap, it means we could not complete read quorum
+                   //   So we will throw an error, which will be eventually retried.
+                   .switchIfEmpty(Flux.defer(() -> {
                        logger.warn("Could not complete read quorum with read quorum value of {}", readQuorumValue);
 
                     return Flux.error(new GoneException(
@@ -634,8 +636,10 @@ private Mono<Boolean> waitForReadBarrierAsync(
                            }
 
                             return Flux.empty();
-                       })).
-                       concatWith(
+                       }))
+                   //   In case the above flux returns empty (which it will after all the retries have been exhausted),
+                   //   We will just return false
+                   .switchIfEmpty(
                                 Flux.defer(() -> {
                                logger.debug("QuorumReader: waitForReadBarrierAsync - TargetGlobalCommittedLsn: {}, MaxGlobalCommittedLsn: {}.", targetGlobalCommittedLSN, maxGlobalCommittedLsn);
                                     return Flux.just(false);