ARTEMIS-5733 Core bridge can lock on flow control

Author: AntonRoskvist

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/AsynchronousProducerCreditsImpl.java

@@ -66,7 +66,7 @@ public void receiveCredits(int credits) {
          }
          callback.onCreditsFlow(balance <= 0, this);
 
-         if (balance < 0 && arriving == 0) {
+         if (balance <= 0 && arriving == 0) {
             // there are no more credits arriving and we are still negative, async large message send asked too much and we need to counter balance
             logger.debug("Starve credits counter balance");
             int request = -balance + windowSize * 2;

artemis-core-client/src/test/java/org/apache/activemq/artemis/core/client/impl/AsynchronousProducerCreditsImplTest.java

@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.core.client.impl;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.mockito.Mockito;
+
+public class AsynchronousProducerCreditsImplTest {
+
+   @Test
+   @Timeout(10)
+   public void testZeroCredits() throws Exception {
+      ClientSessionInternal session = Mockito.mock(ClientSessionInternal.class);
+      AsynchronousProducerCreditsImpl producerCredits = new AsynchronousProducerCreditsImpl(session, null, 0, Mockito.mock(ClientProducerFlowCallback.class));
+      producerCredits.receiveCredits(0);
+      Mockito.verify(session).sendProducerCreditsMessage(0, null);
+   }
+
+}