Complete re-assignment if consumer is closed

Author: acogoluegnes

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

@@ -686,8 +686,8 @@ private void assignConsumersToStream(
                 } else {
                   for (SubscriptionTracker affectedSubscription : subscriptions) {
                     ManagerPool subscriptionPool = null;
-                    boolean reassigned = false;
-                    while (!reassigned) {
+                    boolean reassignmentCompleted = false;
+                    while (!reassignmentCompleted) {
                       try {
                         if (affectedSubscription.consumer.isOpen()) {
                           Client.Broker broker = pickBroker(candidates);
@@ -715,11 +715,14 @@ private void assignConsumersToStream(
                               }
                               subscriptionPool.add(
                                   affectedSubscription, offsetSpecification, false);
-                              reassigned = true;
+                              reassignmentCompleted = true;
+                            } else {
+                              reassignmentCompleted = true;
                             }
                           }
                         } else {
                           LOGGER.debug("Not re-assigning consumer because it has been closed");
+                          reassignmentCompleted = true;
                         }
                       } catch (TimeoutStreamException e) {
                         LOGGER.debug(
@@ -740,6 +743,7 @@ private void assignConsumersToStream(
                       } catch (Exception e) {
                         LOGGER.warn(
                             "Error while re-assigning subscription from stream {}", stream, e);
+                        reassignmentCompleted = true;
                       }
                     }
                   }