Improve recovery

Create a locator connection for each provided URI. This way
a connection can take over when the locator node goes down.
This speeds up recovery.

Track scheduled tasks. This is likely to be disabled in
a stable release. Useful to track down unfinished tasks.

Add retry to operations in the consumer coordinator.

Refresh consumer candidate nodes if the re-assignment of
a consumer times out.

This improvements are based on the feedback from the effects
of a rolling restart in K8S using stream-perf-test. Not all
producers and consumers are recovered after all nodes have
been restarted. The changes in this commit mitigates this problem.

Author: acogoluegnes

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

@@ -13,6 +13,8 @@
 // [email protected].
 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);
     }
   }
 

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

@@ -308,8 +308,18 @@ public void write(
         });
 
     ChannelFuture f;
+    String clientConnectionName =
+        parameters.clientProperties == null
+            ? ""
+            : (parameters.clientProperties.containsKey("connection_name")
+                ? parameters.clientProperties.get("connection_name")
+                : "");
     try {
-      LOGGER.debug("Trying to create stream connection to {}:{}", parameters.host, parameters.port);
+      LOGGER.debug(
+          "Trying to create stream connection to {}:{}, with client connection name '{}'",
+          parameters.host,
+          parameters.port,
+          clientConnectionName);
       f = b.connect(parameters.host, parameters.port).sync();
       this.host = parameters.host;
       this.port = parameters.port;
@@ -1113,6 +1123,9 @@ public QueryOffsetResponse queryOffset(String reference, String stream) {
       request.block();
       QueryOffsetResponse response = request.response.get();
       return response;
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
       throw new StreamException(e);
@@ -2284,7 +2297,7 @@ void block() {
         throw new StreamException("Interrupted while waiting for response");
       }
       if (!completed) {
-        throw new StreamException("Could not get response in " + timeout.toMillis() + " ms");
+        throw new TimeoutStreamException("Could not get response in " + timeout.toMillis() + " ms");
       }
     }