Merge pull request #245 from rabbitmq/recovery-improvements

Improve recovery

Author: acogoluegnes

pom.xml

@@ -51,8 +51,8 @@
     <spotless.check.skip>true</spotless.check.skip>
     <slf4j.version>1.7.36</slf4j.version>
     <logback.version>1.2.11</logback.version>
-    <slf4j.perftool.version>2.0.3</slf4j.perftool.version>
-    <logback.perftool.version>1.3.3</logback.perftool.version>
+    <slf4j.perftool.version>2.0.6</slf4j.perftool.version>
+    <logback.perftool.version>1.3.5</logback.perftool.version>
     <netty.version>4.1.86.Final</netty.version>
     <proton-j.version>0.34.0</proton-j.version>
     <metrics.version>4.2.13</metrics.version>

src/main/java/com/rabbitmq/stream/Environment.java

@@ -42,7 +42,7 @@ static EnvironmentBuilder builder() {
               .getConstructor()
               .newInstance();
     } catch (Exception e) {
-      throw new StreamException(e);
+      throw new StreamException("Error while creating stream environment builder", e);
     }
   }
 

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

@@ -13,6 +13,8 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.Utils.namedRunnable;
+
 import com.rabbitmq.stream.BackOffDelayPolicy;
 import java.time.Duration;
 import java.util.concurrent.Callable;
@@ -42,47 +44,56 @@ private AsyncRetry(
     AtomicReference<Runnable> retryableTaskReference = new AtomicReference<>();
     AtomicInteger attempts = new AtomicInteger(0);
     Runnable retryableTask =
-        () -> {
-          if (Thread.currentThread().isInterrupted()) {
-            LOGGER.debug("Task '{}' interrupted, failing future", description);
-            this.completableFuture.completeExceptionally(new CancellationException());
-            return;
-          }
-          try {
-            V result = task.call();
-            LOGGER.debug("Task '{}' succeeded, completing future", description);
-            completableFuture.complete(result);
-          } catch (Exception e) {
-            int attemptCount = attempts.getAndIncrement();
-            if (retry.test(e)) {
-              if (delayPolicy.delay(attemptCount).equals(BackOffDelayPolicy.TIMEOUT)) {
-                LOGGER.debug(
-                    "Retryable attempts for task '{}' timed out, failing future", description);
-                this.completableFuture.completeExceptionally(new RetryTimeoutException());
-              } else {
-                LOGGER.debug(
-                    "Retryable exception ({}) for task '{}', scheduling another attempt",
-                    e.getClass().getSimpleName(),
-                    description);
-                scheduler.schedule(
-                    retryableTaskReference.get(),
-                    delayPolicy.delay(attemptCount).toMillis(),
-                    TimeUnit.MILLISECONDS);
+        namedRunnable(
+            () -> {
+              if (Thread.currentThread().isInterrupted()) {
+                LOGGER.debug("Task '{}' interrupted, failing future", description);
+                this.completableFuture.completeExceptionally(new CancellationException());
+                return;
+              }
+              try {
+                V result = task.call();
+                LOGGER.debug("Task '{}' succeeded, completing future", description);
+                completableFuture.complete(result);
+              } catch (Exception e) {
+                int attemptCount = attempts.getAndIncrement();
+                if (retry.test(e)) {
+                  if (delayPolicy.delay(attemptCount).equals(BackOffDelayPolicy.TIMEOUT)) {
+                    LOGGER.debug(
+                        "Retryable attempts for task '{}' timed out, failing future", description);
+                    this.completableFuture.completeExceptionally(new RetryTimeoutException());
+                  } else {
+                    LOGGER.debug(
+                        "Retryable exception ({}) for task '{}', scheduling another attempt",
+                        e.getClass().getSimpleName(),
+                        description);
+                    schedule(
+                        scheduler, retryableTaskReference.get(), delayPolicy.delay(attemptCount));
+                  }
+                } else {
+                  LOGGER.debug(
+                      "Non-retryable exception for task '{}', failing future", description);
+                  this.completableFuture.completeExceptionally(e);
+                }
               }
-            } else {
-              LOGGER.debug("Non-retryable exception for task '{}', failing future", description);
-              this.completableFuture.completeExceptionally(e);
-            }
-          }
-        };
+            },
+            description);
     retryableTaskReference.set(retryableTask);
     Duration initialDelay = delayPolicy.delay(attempts.getAndIncrement());
     LOGGER.debug("Scheduling task '{}' with policy {}", description, delayPolicy);
     if (initialDelay.isZero()) {
       retryableTask.run();
     } else {
-      scheduler.schedule(
-          retryableTaskReference.get(), initialDelay.toMillis(), TimeUnit.MILLISECONDS);
+      schedule(scheduler, retryableTaskReference.get(), initialDelay);
+    }
+  }
+
+  private static void schedule(
+      ScheduledExecutorService scheduler, Runnable command, Duration delay) {
+    try {
+      scheduler.schedule(command, delay.toMillis(), TimeUnit.MILLISECONDS);
+    } catch (RuntimeException e) {
+      LOGGER.debug("Error while scheduling command", e);
     }
   }