Do not provide credits on subscription closing

This can trigger unnecessary SUBSCRIPTION_ID_DOES_NOT_EXIST
error messages.

Author: acogoluegnes

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

@@ -217,6 +217,7 @@ private static class SubscriptionTracker {
     private volatile long offset;
     private volatile byte subscriptionIdInClient;
     private volatile ClientSubscriptionsManager manager;
+    private volatile boolean closing = false;
 
     private SubscriptionTracker(
         StreamConsumer consumer,
@@ -230,6 +231,7 @@ private SubscriptionTracker(
     }
 
     synchronized void cancel() {
+      this.closing = true;
       if (this.manager != null) {
         LOGGER.debug("Removing consumer from manager");
         this.manager.remove(this);
@@ -238,6 +240,10 @@ synchronized void cancel() {
       }
     }
 
+    boolean isClosing() {
+      return this.closing;
+    }
+
     synchronized void assign(byte subscriptionIdInClient, ClientSubscriptionsManager manager) {
       this.subscriptionIdInClient = subscriptionIdInClient;
       this.manager = manager;
@@ -367,14 +373,24 @@ private ClientSubscriptionsManager(
       IntStream.range(0, maxConsumersByConnection).forEach(i -> subscriptionTrackers.add(null));
       AtomicBoolean clientInitializedInManager = new AtomicBoolean(false);
       ChunkListener chunkListener =
-          (client, subscriptionId, offset, messageCount, dataSize) ->
+          (client, subscriptionId, offset, messageCount, dataSize) -> {
+            SubscriptionTracker subscriptionTracker =
+                subscriptionTrackers.get(subscriptionId & 0xFF);
+            if (subscriptionTracker != null && !subscriptionTracker.isClosing()) {
               client.credit(subscriptionId, 1);
+            } else {
+              LOGGER.debug(
+                  "Could not find stream subscription {} or subscription closing, not providing credits",
+                  subscriptionId & 0xFF);
+            }
+          };
+
       CreditNotification creditNotification =
           (subscriptionId, responseCode) ->
               LOGGER.debug(
                   "Received credit notification for subscription {}: {}",
-                  subscriptionId,
-                  responseCode);
+                  subscriptionId & 0xFF,
+                  Utils.formatConstant(responseCode));
       MessageListener messageListener =
           (subscriptionId, offset, message) -> {
             SubscriptionTracker subscriptionTracker =