Repository: smallrye/smallrye-reactive-messaging
Issue: AMQP - when using multiple Emitters intermitently fails to retrieve credits on startup

**Scenario**
Quarkus application using SRM AMQP 3.15.0
Application uses Emitters to send messages to multiple Queues on the same Azure Service Bus

**Problem**
On startup, SRM silently fails to obtain credits for **some** of the channels.  This is sporadic but frequent.
This does not cause the startup to fail.  It also does not recover.
Attempts to send messages using these Emitters fail.

**Reproducer**
git clone https://bitbucket.org/alexlitovsky1/srm-amqp-credits-bug.git

**Steps**
1. mvn quarkus:dev
2. Watch the output for these lines `DEBUG [io.sma.rea.mes.amqp] (vert.x-eventloop-thread-0) SRMSG16222: Retrieved credits for channel testX: 1000`
3. There should be 3 of them:  `test1`, `test2`, and `test3`.  Most of the time you will only see a subset of those acquire credits.

**Output from an error scenario**
Credits were obtained for test2 and test3.  Not for test1
```
2022-03-11 09:34:06,595 FINE  [org.apa.qpi.pro.eng.imp.SaslImpl] (vert.x-eventloop-thread-0) Handled outcome: SaslImpl [_outcome=PN_SASL_OK, state=PN_SASL_PASS, done=true, role=CLIENT]
2022-03-11 09:34:06,600 FINE  [org.apa.qpi.pro.eng.imp.SaslImpl] (vert.x-eventloop-thread-0) Handled outcome: SaslImpl [_outcome=PN_SASL_OK, state=PN_SASL_PASS, done=true, role=CLIENT]
2022-03-11 09:34:06,601 FINE  [org.apa.qpi.pro.eng.imp.SaslImpl] (vert.x-eventloop-thread-0) Handled outcome: SaslImpl [_outcome=PN_SASL_OK, state=PN_SASL_PASS, done=true, role=CLIENT]
2022-03-11 09:34:06,612 INFO  [io.sma.rea.mes.amqp] (vert.x-eventloop-thread-0) SRMSG16213: Connection with AMQP broker established
2022-03-11 09:34:06,619 INFO  [io.sma.rea.mes.amqp] (vert.x-eventloop-thread-0) SRMSG16213: Connection with AMQP broker established
2022-03-11 09:34:06,623 INFO  [io.sma.rea.mes.amqp] (vert.x-eventloop-thread-0) SRMSG16213: Connection with AMQP broker established
2022-03-11 09:34:06,640 DEBUG [io.sma.rea.mes.amqp] (vert.x-eventloop-thread-0) SRMSG16222: Retrieved credits for channel `test3`: 1000
2022-03-11 09:34:06,641 DEBUG [io.sma.rea.mes.amqp] (vert.x-eventloop-thread-0) SRMSG16222: Retrieved credits for channel `test2`: 1000
2022-03-11 09:34:12,022 ERROR [io.qua.sch.run.SimpleScheduler] (executor-thread-0) Error occured while executing task for trigger IntervalTrigger [id=1_com.alexlitovsky.bugs.amqp.MessageSender_ScheduledInvoker_sendMessage_28ac1eed06c7507f1cfe170b7c9ab14464f2f6dc, interval=60000]: java.lang.RuntimeException: Timeout sending to emitter1
	at com.alexlitovsky.bugs.amqp.MessageSender.sendMessage(MessageSender.java:46)
	at com.alexlitovsky.bugs.amqp.MessageSender.sendMessage(MessageSender.java:36)
	at com.alexlitovsky.bugs.amqp.MessageSender_Subclass.sendMessage$$superforward1(Unknown Source)
	at com.alexlitovsky.bugs.amqp.MessageSender_Subclass$$function$$1.apply(Unknown Source)
	at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
	at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
	at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:49)
	at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(Unknown Source)
	at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
	at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
	at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
	at com.alexlitovsky.bugs.amqp.MessageSender_Subclass.sendMessage(Unknown Source)
	at com.alexlitovsky.bugs.amqp.MessageSender_ClientProxy.sendMessage(Unknown Source)
	at com.alexlitovsky.bugs.amqp.MessageSender_ScheduledInvoker_sendMessage_28ac1eed06c7507f1cfe170b7c9ab14464f2f6dc.invokeBean(Unknown Source)
	at io.quarkus.arc.runtime.BeanInvoker.invoke(BeanInvoker.java:20)
	at io.quarkus.scheduler.runtime.SimpleScheduler$ScheduledTask$1.run(SimpleScheduler.java:274)
	at io.quarkus.vertx.core.runtime.VertxCoreRecorder$13.runWith(VertxCoreRecorder.java:548)
	at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2449)
	at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1478)
	at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)
	at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.util.concurrent.TimeoutException
	at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1886)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2021)
	at com.alexlitovsky.bugs.amqp.MessageSender.sendMessage(MessageSender.java:43)
	... 22 more
```

Changed files:
smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpCreditBasedSender.java
smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpCreditTest.java

diff --git a/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpCreditBasedSender.java b/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpCreditBasedSender.java
index 4f646ddfa..55314c8dc 100644
--- a/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpCreditBasedSender.java
+++ b/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpCreditBasedSender.java
@@ -56,6 +56,12 @@ public class AmqpCreditBasedSender implements Processor<Message<?>, Message<?>>,
 
     private volatile boolean isAnonymous;
 
+    /**
+     * A flag tracking if we are retrieving the credits for the sender.
+     * It avoids flooding the broker with credit requests.
+     */
+    private volatile boolean creditRetrievalInProgress = false;
+
     public AmqpCreditBasedSender(AmqpConnector connector, ConnectionHolder holder,
             AmqpConnectorOutgoingConfiguration configuration, Uni<AmqpSender> retrieveSender) {
         this.connector = connector;
@@ -152,6 +158,9 @@ public class AmqpCreditBasedSender implements Processor<Message<?>, Message<?>>,
             subscription.request(credits);
             return credits;
         }
+        if (credits == 0L && subscription != Subscriptions.CANCELLED) {
+            onNoMoreCredit(sender);
+        }
         return 0L;
     }
 
@@ -188,22 +197,28 @@ public class AmqpCreditBasedSender implements Processor<Message<?>, Message<?>>,
     }
 
     private void onNoMoreCredit(AmqpSender sender) {
-        log.noMoreCreditsForChannel(configuration.getChannel());
-        holder.getContext().runOnContext(() -> {
-            if (isCancelled()) {
-                return;
-            }
-            long c = setCreditsAndRequest(sender);
-            if (c == 0L) { // still no credits, schedule a periodic retry
-                holder.getVertx().setPeriodic(configuration.getCreditRetrievalPeriod(), id -> {
-                    if (setCreditsAndRequest(sender) != 0L || isCancelled()) {
-                        // Got our new credits or the application has been terminated,
-                        // we cancel the periodic task.
-                        holder.getVertx().cancelTimer(id);
-                    }
-                });
-            }
-        });
+        if (!creditRetrievalInProgress) {
+            creditRetrievalInProgress = true;
+            log.noMoreCreditsForChannel(configuration.getChannel());
+            holder.getContext().runOnContext(() -> {
+                if (isCancelled()) {
+                    return;
+                }
+                long c = setCreditsAndRequest(sender);
+                if (c == 0L) { // still no credits, schedule a periodic retry
+                    holder.getVertx().setPeriodic(configuration.getCreditRetrievalPeriod(), id -> {
+                        if (setCreditsAndRequest(sender) != 0L || isCancelled()) {
+                            // Got our new credits or the application has been terminated,
+                            // we cancel the periodic task.
+                            holder.getVertx().cancelTimer(id);
+                            creditRetrievalInProgress = false;
+                        }
+                    });
+                } else {
+                    creditRetrievalInProgress = false;
+                }
+            });
+        }
     }
 
     private boolean isCancelled() {
@@ -236,7 +251,9 @@ public class AmqpCreditBasedSender implements Processor<Message<?>, Message<?>>,
             getSenderAndCredits()
                     .onItem().ignore().andContinueWithNull()
                     .subscribe().with(s -> {
-                    }, f -> downstream.get().onError(f));
+                    }, f -> {
+                        downstream.get().onError(f);
+                    });
         }
     }
 
diff --git a/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpCreditTest.java b/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpCreditTest.java
index 0a207cebd..bc24f00be 100644
--- a/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpCreditTest.java
+++ b/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpCreditTest.java
@@ -60,7 +60,7 @@ public class AmqpCreditTest extends AmqpTestBase {
 
         assertThat(msgsReceived.await(20, TimeUnit.SECONDS))
                 .withFailMessage("Sent %s msgs but %s remain outstanding", msgCount, msgsReceived.getCount()).isTrue();
-        List<Integer> expectedPayloads = IntStream.range(0, msgCount).mapToObj(Integer::valueOf).collect(Collectors.toList());
+        List<Integer> expectedPayloads = IntStream.range(0, msgCount).boxed().collect(Collectors.toList());
         assertThat(payloadsReceived).containsAll(expectedPayloads);
     }