[Service Bus] Allow 0 prefetch and dynamically use batch size to request link credits

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusProcessorClient.java

@@ -150,8 +150,8 @@ private synchronized void receiveMessages() {
         }
         ServiceBusReceiverAsyncClient receiverClient = asyncClient.get();
         receiverClient.receiveMessagesWithContext()
-            .parallel(processorOptions.getMaxConcurrentCalls())
-            .runOn(Schedulers.boundedElastic())
+            .parallel(processorOptions.getMaxConcurrentCalls(), 1)
+            .runOn(Schedulers.boundedElastic(), 1)
             .subscribe(new Subscriber<ServiceBusMessageContext>() {
                 @Override
                 public void onSubscribe(Subscription subscription) {

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java

@@ -580,7 +580,7 @@ public Flux<ServiceBusReceivedMessage> receiveMessages() {
         // to auto-refill the prefetch buffer. A request will retrieve one message from this buffer.
         // If receiverOptions.prefetchCount is 0 (default value),
         // the request will add a link credit so one message is retrieved from the service.
-        return receiveMessagesNoBackPressure().limitRate(1);
+        return receiveMessagesNoBackPressure().limitRate(1, 0);
     }
 
     Flux<ServiceBusReceivedMessage> receiveMessagesNoBackPressure() {

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReceiveLinkProcessor.java

@@ -567,6 +567,13 @@ private int getCreditsToAdd(int linkCredits) {
             if (r <= Integer.MAX_VALUE) {
                 expectedTotalCredit = (int) r;
             } else {
+                //This won't really happen in reality.
+                //For async client, receiveMessages() calls "return receiveMessagesNoBackPressure().limitRate(1, 0);".
+                //So it will request one by one from this link processor, even though the user's request has no
+                //back pressure.
+                //For sync client, the sync subscriber has back pressure.
+                //The request count uses the the argument of method receiveMessages(int maxMessages).
+                //It's at most Integer.MAX_VALUE.
                 expectedTotalCredit = Integer.MAX_VALUE;
             }
         } else {