[fix][broker] Restore the behavior to dispatch batch messages according to consumer permits (#24092)

Author: BewareMyPower

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -843,14 +843,14 @@ protected synchronized boolean trySendMessagesToConsumers(ReadType readType, Lis
                         c, c.getAvailablePermits());
             }
 
-            int maxMessagesInThisBatch =
-                    Math.max(remainingMessages, serviceConfig.getDispatcherMaxRoundRobinBatchSize());
+            int maxMessagesInThisBatch = Math.min(remainingMessages, availablePermits);
             if (c.getMaxUnackedMessages() > 0) {
                 // Calculate the maximum number of additional unacked messages allowed
                 int maxAdditionalUnackedMessages = Math.max(c.getMaxUnackedMessages() - c.getUnackedMessages(), 0);
                 maxMessagesInThisBatch = Math.min(maxMessagesInThisBatch, maxAdditionalUnackedMessages);
             }
-            int maxEntriesInThisBatch = Math.min(availablePermits,
+            // TODO: add tests to verify dispatcherMaxRoundRobinBatchSize is respected
+            int maxEntriesInThisBatch = Math.min(serviceConfig.getDispatcherMaxRoundRobinBatchSize(),
                             // use the average batch size per message to calculate the number of entries to
                             // dispatch. round up to the next integer without using floating point arithmetic.
                             (maxMessagesInThisBatch + avgBatchSizePerMsg - 1) / avgBatchSizePerMsg);

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java

@@ -1012,6 +1012,10 @@ public void testBatchMessageDispatchingAccordingToPermits() throws Exception {
         }
         FutureUtil.waitForAll(sendFutureList).get();
 
+        Awaitility.await().atMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+            assertTrue(consumer1.numMessagesInQueue() > 0);
+            assertTrue(consumer2.numMessagesInQueue() > 0);
+        });
         assertEquals(consumer1.numMessagesInQueue(), batchMessages, batchMessages);
         assertEquals(consumer2.numMessagesInQueue(), batchMessages, batchMessages);