From 859a4a98798d2c2b39449acdba5d9570976bee46 Mon Sep 17 00:00:00 2001
From: Sergey Novikov <asn100@motorolasolutions.com>
Date: Wed, 18 Nov 2020 23:08:00 +0100
Subject: [PATCH] deprecated retryWhen usage substituted

---
 .../java/com/azure/core/amqp/implementation/RetryUtil.java   | 5 +++--
 .../com/azure/cosmos/implementation/BackoffRetryUtility.java | 5 +++--
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java
index 2eb65adc3d4df..cc84c04be68b7 100644
--- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java
+++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java
@@ -12,6 +12,7 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.util.Locale;
@@ -55,7 +56,7 @@ public static AmqpRetryPolicy getRetryPolicy(AmqpRetryOptions options) {
      */
     public static <T> Flux<T> withRetry(Flux<T> source, Duration operationTimeout, AmqpRetryPolicy retryPolicy) {
         return Flux.defer(() -> source.timeout(operationTimeout))
-            .retryWhen(errors -> retry(errors, retryPolicy));
+            .retryWhen(Retry.withThrowable(errors -> retry(errors, retryPolicy)));
     }
 
     /**
@@ -67,7 +68,7 @@ public static <T> Flux<T> withRetry(Flux<T> source, Duration operationTimeout, A
      */
     public static <T> Mono<T> withRetry(Mono<T> source, Duration operationTimeout, AmqpRetryPolicy retryPolicy) {
         return Mono.defer(() -> source.timeout(operationTimeout))
-            .retryWhen(errors -> retry(errors, retryPolicy));
+            .retryWhen(Retry.withThrowable(errors -> retry(errors, retryPolicy)));
     }
 
     private static Flux<Long> retry(Flux<Throwable> source, AmqpRetryPolicy retryPolicy) {
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java
index 9426daa4422e1..e39c729d118c2 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java
@@ -5,6 +5,7 @@
 import com.azure.cosmos.implementation.directconnectivity.AddressSelector;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.util.concurrent.Callable;
@@ -37,7 +38,7 @@ static public <T> Mono<T> executeRetry(Callable<Mono<T>> callbackMethod,
             } catch (Exception e) {
                 return Mono.error(e);
             }
-        }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy));
+        }).retryWhen(Retry.withThrowable(RetryUtils.toRetryWhenFunc(retryPolicy)));
     }
 
     static public <T> Flux<T> fluxExecuteRetry(Callable<Flux<T>> callbackMethod, IRetryPolicy retryPolicy) {
@@ -48,7 +49,7 @@ static public <T> Flux<T> fluxExecuteRetry(Callable<Flux<T>> callbackMethod, IRe
             } catch (Exception e) {
                 return Flux.error(e);
             }
-        }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy));
+        }).retryWhen(Retry.withThrowable(RetryUtils.toRetryWhenFunc(retryPolicy)));
     }
 
     static public <T> Mono<T> executeAsync(