Increase number of locator lookup attempts

Author: acogoluegnes

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

@@ -2824,7 +2824,17 @@ public void channelInactive(ChannelHandlerContext ctx) {
       // because it will be handled later anyway.
       if (shutdownReason == null) {
         if (closing.compareAndSet(false, true)) {
-          executorService.submit(() -> closingSequence(ShutdownReason.UNKNOWN));
+          if (executorService == null) {
+            // the TCP connection is closed before the state is initialized
+            // we do our best the execute the closing sequence
+            new Thread(
+                    () -> {
+                      closingSequence(ShutdownReason.UNKNOWN);
+                    })
+                .start();
+          } else {
+            executorService.submit(() -> closingSequence(ShutdownReason.UNKNOWN));
+          }
         }
       }
     }

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

@@ -742,7 +742,7 @@ static <T> T locatorOperation(
       Function<Client, T> operation,
       Supplier<Client> clientSupplier,
       BackOffDelayPolicy backOffDelayPolicy) {
-    int maxAttempt = 3;
+    int maxAttempt = 5;
     int attempt = 0;
     boolean executed = false;
     Exception lastException = null;

src/test/java/com/rabbitmq/stream/impl/RecoveryClusterTest.java

@@ -23,6 +23,7 @@
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import ch.qos.logback.classic.Level;
 import com.google.common.util.concurrent.RateLimiter;
 import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.impl.TestUtils.Sync;
@@ -57,10 +58,13 @@ public class RecoveryClusterTest {
   TestInfo testInfo;
   EventLoopGroup eventLoopGroup;
   EnvironmentBuilder environmentBuilder;
+  static Level producersCoordinatorLogLevel;
 
   @BeforeAll
   static void initAll() {
     nodes = Cli.nodes();
+    producersCoordinatorLogLevel =
+        TestUtils.newLoggerLevel(ProducersCoordinator.class, Level.DEBUG);
   }
 
   @BeforeEach
@@ -82,6 +86,13 @@ void tearDown() {
     }
   }
 
+  @AfterAll
+  static void tearDownAll() {
+    if (producersCoordinatorLogLevel != null) {
+      TestUtils.newLoggerLevel(ProducersCoordinator.class, producersCoordinatorLogLevel);
+    }
+  }
+
   @ParameterizedTest
   @CsvSource({
     "false,false",

src/test/resources/logback-test.xml

@@ -7,6 +7,7 @@
 
     <logger name="com.rabbitmq.stream" level="warn" />
     <logger name="com.rabbitmq.stream.impl.Utils" level="warn" />
+    <logger name="com.rabbitmq.stream.impl.StreamEnvironment" level="warn" />
     <logger name="com.rabbitmq.stream.impl.ConsumersCoordinator" level="warn" />
     <logger name="com.rabbitmq.stream.impl.ProducersCoordinator" level="warn" />
     <logger name="com.rabbitmq.stream.impl.RecoveryClusterTest" level="info" />