Base retry completion on backoff delay policy

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/ConsumersCoordinator.java

@@ -16,8 +16,10 @@
 import static com.rabbitmq.stream.impl.Utils.convertCodeToException;
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
 import static com.rabbitmq.stream.impl.Utils.isSac;
+import static com.rabbitmq.stream.impl.Utils.jsonField;
 import static com.rabbitmq.stream.impl.Utils.namedFunction;
 import static com.rabbitmq.stream.impl.Utils.namedRunnable;
+import static com.rabbitmq.stream.impl.Utils.quote;
 
 import com.rabbitmq.stream.BackOffDelayPolicy;
 import com.rabbitmq.stream.Constants;
@@ -224,7 +226,6 @@ private void addToManager(
     }
   }
 
-  // for testing
   int managerCount() {
     return this.managers.size();
   }
@@ -297,22 +298,6 @@ public void close() {
     }
   }
 
-  private static String quote(String value) {
-    if (value == null) {
-      return "null";
-    } else {
-      return "\"" + value + "\"";
-    }
-  }
-
-  private static String jsonField(String name, Number value) {
-    return quote(name) + " : " + value.longValue();
-  }
-
-  private static String jsonField(String name, String value) {
-    return quote(name) + " : " + (value == null ? "null" : quote(value));
-  }
-
   @Override
   public String toString() {
     StringBuilder builder = new StringBuilder("{");
@@ -768,17 +753,7 @@ private void assignConsumersToStream(
                   consumersClosingCallback.run();
                 } else {
                   for (SubscriptionTracker affectedSubscription : subscriptions) {
-                    if (affectedSubscription.compareAndSet(
-                        SubscriptionState.ACTIVE, SubscriptionState.RECOVERING)) {
-                      recoverSubscription(candidates, affectedSubscription);
-                    } else {
-                      LOGGER.debug(
-                          "Not recovering consumer {} from stream {}, state is {}, expected is {}",
-                          affectedSubscription.consumer.id(),
-                          affectedSubscription.stream,
-                          affectedSubscription.state(),
-                          SubscriptionState.ACTIVE);
-                    }
+                    maybeRecoverSubscription(candidates, affectedSubscription);
                   }
                   if (maybeCloseClient) {
                     this.closeIfEmpty();
@@ -800,6 +775,27 @@ private void assignConsumersToStream(
               });
     }
 
+    private void maybeRecoverSubscription(List<Broker> candidates, SubscriptionTracker tracker) {
+      if (tracker.compareAndSet(SubscriptionState.ACTIVE, SubscriptionState.RECOVERING)) {
+        try {
+          recoverSubscription(candidates, tracker);
+        } catch (Exception e) {
+          LOGGER.warn(
+              "Error while recovering consumer {} from stream '{}'. Reason: {}",
+              tracker.consumer.id(),
+              tracker.stream,
+              Utils.exceptionMessage(e));
+        }
+      } else {
+        LOGGER.debug(
+            "Not recovering consumer {} from stream {}, state is {}, expected is {}",
+            tracker.consumer.id(),
+            tracker.stream,
+            tracker.state(),
+            SubscriptionState.ACTIVE);
+      }
+    }
+
     private void recoverSubscription(List<Broker> candidates, SubscriptionTracker tracker) {
       boolean reassignmentCompleted = false;
       while (!reassignmentCompleted) {

src/main/java/com/rabbitmq/stream/impl/ProducersCoordinator.java

@@ -15,8 +15,10 @@
 
 import static com.rabbitmq.stream.impl.Utils.callAndMaybeRetry;
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
+import static com.rabbitmq.stream.impl.Utils.jsonField;
 import static com.rabbitmq.stream.impl.Utils.namedFunction;
 import static com.rabbitmq.stream.impl.Utils.namedRunnable;
+import static com.rabbitmq.stream.impl.Utils.quote;
 import static java.util.stream.Collectors.toSet;
 
 import com.rabbitmq.stream.BackOffDelayPolicy;
@@ -238,22 +240,6 @@ int nodesConnected() {
     return this.managers.stream().map(m -> m.name).collect(toSet()).size();
   }
 
-  private static String quote(String value) {
-    if (value == null) {
-      return "null";
-    } else {
-      return "\"" + value + "\"";
-    }
-  }
-
-  private static String jsonField(String name, Number value) {
-    return quote(name) + " : " + value.longValue();
-  }
-
-  private static String jsonField(String name, String value) {
-    return quote(name) + " : " + quote(value);
-  }
-
   @Override
   public String toString() {
     StringBuilder builder = new StringBuilder("{");
@@ -700,17 +686,7 @@ private void assignProducersToNewManagers(
               broker -> {
                 String key = keyForNode(broker);
                 LOGGER.debug("Assigning {} producer(s) to {}", trackers.size(), key);
-                trackers.forEach(
-                    tracker -> {
-                      if (tracker.markRecoveryInProgress()) {
-                        recoverAgent(broker, tracker);
-                      } else {
-                        LOGGER.debug(
-                            "Not recovering {} (stream '{}'), recovery is already is progress",
-                            tracker.type(),
-                            tracker.stream());
-                      }
-                    });
+                trackers.forEach(tracker -> maybeRecoverAgent(broker, tracker));
               })
           .exceptionally(
               ex -> {
@@ -735,6 +711,26 @@ private void assignProducersToNewManagers(
               });
     }
 
+    private void maybeRecoverAgent(Broker broker, AgentTracker tracker) {
+      if (tracker.markRecoveryInProgress()) {
+        try {
+          recoverAgent(broker, tracker);
+        } catch (Exception e) {
+          LOGGER.warn(
+              "Error while recovering {} tracker {} (stream '{}'). Reason: {}",
+              tracker.type(),
+              tracker.uniqueId(),
+              tracker.stream(),
+              Utils.exceptionMessage(e));
+        }
+      } else {
+        LOGGER.debug(
+            "Not recovering {} (stream '{}'), recovery is already is progress",
+            tracker.type(),
+            tracker.stream());
+      }
+    }
+
     private void recoverAgent(Broker node, AgentTracker tracker) {
       boolean reassignmentCompleted = false;
       while (!reassignmentCompleted) {

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -812,6 +812,10 @@ public String toString() {
                 })
             .collect(Collectors.joining(","))
         + "], "
+        + Utils.jsonField("producer_client_count", this.producersCoordinator.clientCount())
+        + ","
+        + Utils.jsonField("consumer_client_count", this.consumersCoordinator.managerCount())
+        + ","
         + "\"producers\" : "
         + this.producersCoordinator
         + ", \"consumers\" : "

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -165,7 +165,12 @@ static <T, R> Function<T, R> namedFunction(Function<T, R> task, String format, O
 
   static <T> T callAndMaybeRetry(
       Supplier<T> operation, Predicate<Exception> retryCondition, String format, Object... args) {
-    return callAndMaybeRetry(operation, retryCondition, i -> Duration.ZERO, format, args);
+    return callAndMaybeRetry(
+        operation,
+        retryCondition,
+        i -> i >= 3 ? BackOffDelayPolicy.TIMEOUT : Duration.ZERO,
+        format,
+        args);
   }
 
   static <T> T callAndMaybeRetry(
@@ -177,32 +182,38 @@ static <T> T callAndMaybeRetry(
     String description = format(format, args);
     int attempt = 0;
     Exception lastException = null;
-    while (attempt++ < 3) {
+    boolean keepTrying = true;
+    while (keepTrying) {
       try {
-        return operation.get();
+        attempt++;
+        T result = operation.get();
+        LOGGER.debug("Operation '{}' completed after {} attempt(s)", description, attempt);
+        return result;
       } catch (Exception e) {
         lastException = e;
         if (retryCondition.test(e)) {
           LOGGER.debug("Operation '{}' failed, retrying...", description);
-          Duration delay = delayPolicy.delay(attempt - 1);
-          if (!delay.isZero()) {
+          Duration delay = delayPolicy.delay(attempt);
+          if (BackOffDelayPolicy.TIMEOUT.equals(delay)) {
+            keepTrying = false;
+          } else if (!delay.isZero()) {
             try {
               Thread.sleep(delay.toMillis());
             } catch (InterruptedException ex) {
               Thread.interrupted();
               lastException = ex;
-              break;
+              keepTrying = false;
             }
           }
         } else {
-          break;
+          keepTrying = false;
         }
       }
     }
     String message =
         format(
-            "Could not complete task '%s' after %d attempt(s) (reason: {})",
-            description, --attempt, exceptionMessage(lastException));
+            "Could not complete task '%s' after %d attempt(s) (reason: %s)",
+            description, attempt, exceptionMessage(lastException));
     LOGGER.debug(message);
     if (lastException == null) {
       throw new StreamException(message);
@@ -455,4 +466,20 @@ public String toString() {
       return this.name;
     }
   }
+
+  static String quote(String value) {
+    if (value == null) {
+      return "null";
+    } else {
+      return "\"" + value + "\"";
+    }
+  }
+
+  static String jsonField(String name, Number value) {
+    return quote(name) + " : " + value.longValue();
+  }
+
+  static String jsonField(String name, String value) {
+    return quote(name) + " : " + quote(value);
+  }
 }