Improve recovery

Extra debug logging, adapt recovery delay for cluster tests.

Author: acogoluegnes

ci/start-broker.sh

@@ -14,6 +14,7 @@ wait_for_message() {
 
 make -C "${PWD}"/tls-gen/basic
 
+rm -rf rabbitmq-configuration
 mkdir -p rabbitmq-configuration/tls
 cp -R "${PWD}"/tls-gen/basic/result/* rabbitmq-configuration/tls
 chmod o+r rabbitmq-configuration/tls/*

pom.xml

@@ -78,6 +78,7 @@
     <maven.gpg.plugin.version>3.2.7</maven.gpg.plugin.version>
     <buildnumber.plugin.version>3.2.1</buildnumber.plugin.version>
     <maven-resources-plugin.version>3.3.1</maven-resources-plugin.version>
+    <maven-clean-plugin.version>3.4.0</maven-clean-plugin.version>
     <maven-source-plugin.version>3.3.1</maven-source-plugin.version>
     <maven-javadoc-plugin.version>3.11.2</maven-javadoc-plugin.version>
     <maven.jar.plugin.version>3.4.2</maven.jar.plugin.version>
@@ -100,7 +101,6 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <maven.packagecloud.wagon.version>0.0.6</maven.packagecloud.wagon.version>
     <nexus-staging-maven-plugin.version>1.7.0</nexus-staging-maven-plugin.version>
-
     <maven.javadoc.skip>true</maven.javadoc.skip>
     <gpg.skip>true</gpg.skip>
   </properties>
@@ -356,6 +356,12 @@
         </configuration>
       </plugin>
 
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-clean-plugin</artifactId>
+        <version>${maven-clean-plugin.version}</version>
+      </plugin>
+
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-dependency-plugin</artifactId>

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2023 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2024 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -14,6 +14,7 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.ThreadUtils.isVirtual;
 import static com.rabbitmq.stream.impl.Utils.namedRunnable;
 
 import com.rabbitmq.stream.BackOffDelayPolicy;
@@ -53,11 +54,19 @@ private AsyncRetry(
                 return;
               }
               try {
+                LOGGER.debug(
+                    "Running task '{}' (virtual threads: {})",
+                    description,
+                    isVirtual(Thread.currentThread()));
                 V result = task.call();
                 LOGGER.debug("Task '{}' succeeded, completing future", description);
                 completableFuture.complete(result);
               } catch (Exception e) {
                 int attemptCount = attempts.getAndIncrement();
+                LOGGER.debug(
+                    "Attempt {} for task '{}' failed, checking retry policy",
+                    attemptCount,
+                    description);
                 if (retry.test(e)) {
                   if (delayPolicy.delay(attemptCount).equals(BackOffDelayPolicy.TIMEOUT)) {
                     LOGGER.debug(

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

@@ -15,6 +15,7 @@
 package com.rabbitmq.stream.impl;
 
 import static com.rabbitmq.stream.Constants.*;
+import static com.rabbitmq.stream.impl.ThreadUtils.threadFactory;
 import static com.rabbitmq.stream.impl.Utils.DEFAULT_USERNAME;
 import static com.rabbitmq.stream.impl.Utils.encodeRequestCode;
 import static com.rabbitmq.stream.impl.Utils.encodeResponseCode;
@@ -45,7 +46,6 @@
 import com.rabbitmq.stream.impl.Client.ShutdownContext.ShutdownReason;
 import com.rabbitmq.stream.impl.ServerFrameHandler.FrameHandler;
 import com.rabbitmq.stream.impl.ServerFrameHandler.FrameHandlerInfo;
-import com.rabbitmq.stream.impl.Utils.NamedThreadFactory;
 import com.rabbitmq.stream.metrics.MetricsCollector;
 import com.rabbitmq.stream.metrics.NoOpMetricsCollector;
 import com.rabbitmq.stream.sasl.CredentialsProvider;
@@ -164,7 +164,7 @@ public class Client implements AutoCloseable {
   final CompressionCodecFactory compressionCodecFactory;
   private final Consumer<ShutdownContext.ShutdownReason> shutdownListenerCallback;
   private final ToLongFunction<Object> publishSequenceFunction =
-      new ToLongFunction<Object>() {
+      new ToLongFunction<>() {
         private final AtomicLong publishSequence = new AtomicLong(0);
 
         @Override
@@ -302,6 +302,7 @@ public void initChannel(SocketChannel ch) {
           }
         });
 
+    this.nettyClosing = Utils.makeIdempotent(this::closeNetty);
     ChannelFuture f;
     String clientConnectionName = parameters.clientProperties.getOrDefault("connection_name", "");
     try {
@@ -326,13 +327,11 @@ public void initChannel(SocketChannel ch) {
         throw new StreamException(message, e);
       }
     }
-
     this.channel = f.channel();
-    this.nettyClosing = Utils.makeIdempotent(this::closeNetty);
     ExecutorServiceFactory executorServiceFactory = parameters.executorServiceFactory;
     if (executorServiceFactory == null) {
       this.executorService =
-          Executors.newSingleThreadExecutor(new NamedThreadFactory(clientConnectionName + "-"));
+          Executors.newSingleThreadExecutor(threadFactory(clientConnectionName + "-"));
     } else {
       this.executorService = executorServiceFactory.get();
     }
@@ -341,7 +340,7 @@ public void initChannel(SocketChannel ch) {
     if (dispatchingExecutorServiceFactory == null) {
       this.dispatchingExecutorService =
           Executors.newSingleThreadExecutor(
-              new NamedThreadFactory("dispatching-" + clientConnectionName + "-"));
+              threadFactory("dispatching-" + clientConnectionName + "-"));
     } else {
       this.dispatchingExecutorService = dispatchingExecutorServiceFactory.get();
     }
@@ -1443,7 +1442,7 @@ void closingSequence(ShutdownContext.ShutdownReason reason) {
 
   private void closeNetty() {
     try {
-      if (this.channel.isOpen()) {
+      if (this.channel != null && this.channel.isOpen()) {
         LOGGER.debug("Closing Netty channel");
         this.channel.close().get(10, TimeUnit.SECONDS);
       }