[EH Perf improvement] Translate Reactor's no-backpressure to prefetch in amqp Flow (Azure#22819)

* Translate Reactor's no-backpressure to prefetch in amqp Flow

* Clarified the credit for no-backpressure case in the getCreditsToAdd(..) method javadoc

Author: anuchandy

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/AmqpReceiveLinkProcessor.java

@@ -572,7 +572,8 @@ private void addCreditsToLink(String message) {
 
     /**
      * Gets the number of credits to add based on {@link #requested} and how many messages are still in queue.
-     * If {@link #requested} is {@link Long#MAX_VALUE}, then we add credits 1 by 1. Similar to Track 1's behaviour.
+     * If {@link #requested} is {@link Long#MAX_VALUE}, which indicates no-backpressure,
+     * then we use the {@link #prefetch} value as credit.
      *
      * @return The number of credits to add.
      */
@@ -584,7 +585,7 @@ private int getCreditsToAdd() {
         if (subscriber == null || request == 0) {
             credits = 0;
         } else if (request == Long.MAX_VALUE) {
-            credits = 1;
+            credits = prefetch;
         } else {
             final int remaining = Long.valueOf(request).intValue() - messageQueue.size();
             credits = Math.max(remaining, 0);

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/implementation/AmqpReceiveLinkProcessorTest.java

@@ -153,8 +153,8 @@ void createNewLink() {
         Assertions.assertNotNull(value);
 
         final Integer creditValue = value.get();
-        // Expecting 1 because it is Long.MAX_VALUE.
-        Assertions.assertEquals(1, creditValue);
+        // Expecting PREFETCH because it is Long.MAX_VALUE.
+        Assertions.assertEquals(PREFETCH, creditValue);
     }
 
     /**
@@ -462,8 +462,8 @@ void receivesUntilFirstLinkClosed() {
         Assertions.assertNotNull(value);
 
         final Integer creditValue = value.get();
-        // Expecting 1 because it is Long.MAX_VALUE.
-        Assertions.assertEquals(1, creditValue);
+        // Expecting PREFETCH because it is Long.MAX_VALUE.
+        Assertions.assertEquals(PREFETCH, creditValue);
     }
 
     @Test
@@ -496,8 +496,8 @@ void receivesFromFirstLink() {
         Assertions.assertNotNull(value);
 
         final Integer creditValue = value.get();
-        // Expecting 1 because it is Long.MAX_VALUE.
-        Assertions.assertEquals(1, creditValue);
+        // Expecting PREFETCH because it is Long.MAX_VALUE.
+        Assertions.assertEquals(PREFETCH, creditValue);
     }
 
     /**