fix on failed emission attempts

Azure · Dec 8, 2024 · e8bc112 · e8bc112
1 parent 24590b6
commit e8bc112
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/data/src/main/java/com/microsoft/azure/kusto/data/auth/CloudInfo.java b/data/src/main/java/com/microsoft/azure/kusto/data/auth/CloudInfo.java
@@ -110,7 +110,10 @@ public Mono<Void> initiateCloudInfoRetrieval(String clusterUrl, @Nullable HttpCl
         return Mono.fromCallable(() -> UriUtils.setPathForUri(clusterUrl, ""))
                 .map(url -> cache.computeIfAbsent(url, key -> fetchCloudInfoAsync(url, givenHttpClient)
                         .retryWhen(new ExponentialRetry<>(exponentialRetryTemplate).retry())
-                        .doOnSuccess(cloudInfo -> sink.emitValue(cloudInfo, Sinks.EmitFailureHandler.FAIL_FAST))
+
+                        // Since we are trying to achieve concurrent emission to multiple subscribers (Sinks perform emissions serially),
+                        // a handler is specified in order to retry emission attempts for up to the specified duration
+                        .doOnSuccess(cloudInfo -> sink.emitValue(cloudInfo, Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(5))))
                         .onErrorMap(e -> ExceptionUtils.unwrapCloudInfoException(url, e))))
                 .then();
     }