Stop reconnecting locator is environment is closed

Author: acogoluegnes

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

@@ -356,13 +356,19 @@ class StreamEnvironment implements Environment {
                           clientFactory,
                           this.locatorReconnectionScheduledExecutorService,
                           this.recoveryBackOffDelayPolicy,
-                          l.label());
+                          l.label(),
+                          this.closed);
                     }
                   });
             }
           };
       if (lazyInit) {
-        this.locatorInitializationSequence = locatorInitSequence;
+        this.locatorInitializationSequence =
+            () -> {
+              if (!this.closed.get()) {
+                locatorInitSequence.run();
+              }
+            };
       } else {
         locatorInitSequence.run();
         locatorsInitialized.set(true);
@@ -391,7 +397,7 @@ private ShutdownListener shutdownListener(
         shutdownContext -> {
           String label = locator.label();
           LOGGER.debug("Locator {} disconnected", label);
-          if (shutdownContext.isShutdownUnexpected()) {
+          if (shutdownContext.isShutdownUnexpected() && !this.closed.get()) {
             locator.client(null);
             BackOffDelayPolicy delayPolicy = recoveryBackOffDelayPolicy;
             LOGGER.debug(
@@ -408,7 +414,8 @@ private ShutdownListener shutdownListener(
                 clientFactory,
                 this.locatorReconnectionScheduledExecutorService,
                 delayPolicy,
-                label);
+                label,
+                this.closed);
           } else {
             LOGGER.debug("Locator connection '{}' closing normally", label);
           }
@@ -425,7 +432,8 @@ private static void scheduleLocatorConnection(
       Function<Client.ClientParameters, Client> clientFactory,
       ScheduledExecutorService scheduler,
       BackOffDelayPolicy delayPolicy,
-      String locatorLabel) {
+      String locatorLabel,
+      AtomicBoolean closed) {
     LOGGER.debug(
         "Scheduling locator '{}' connection with delay policy {}", locatorLabel, delayPolicy);
     try {
@@ -452,6 +460,7 @@ private static void scheduleLocatorConnection(
           .description("Locator '%s' connection", locatorLabel)
           .scheduler(scheduler)
           .delayPolicy(delayPolicy)
+          .retry(ignored -> !closed.get())
           .build()
           .thenAccept(locator::client)
           .exceptionally(

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

@@ -53,10 +53,8 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
-@ExtendWith({
-  TestUtils.StreamTestInfrastructureExtension.class,
-  BrokerVersionAtLeast311Condition.class
-})
+@ExtendWith({BrokerVersionAtLeast311Condition.class})
+@StreamTestInfrastructure
 @SingleActiveConsumer
 public class SacSuperStreamConsumerTest {
   EventLoopGroup eventLoopGroup;