[improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (apache#24386)

Author: sandeep-mst

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java

@@ -56,7 +56,7 @@ public NonPersistentDispatcherSingleActiveConsumer(SubType subscriptionType, int
     }
 
     @Override
-    public void sendMessages(List<Entry> entries) {
+    public synchronized void sendMessages(List<Entry> entries) {
         Consumer currentConsumer = getActiveConsumer();
         if (currentConsumer != null && currentConsumer.getAvailablePermits() > 0 && currentConsumer.isWritable()) {
             SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java

@@ -135,7 +135,7 @@ protected Map<Consumer, List<Integer>> initialValue() throws Exception {
             };
 
     @Override
-    public void sendMessages(List<Entry> entries) {
+    public synchronized void sendMessages(List<Entry> entries) {
         if (entries.isEmpty()) {
             return;
         }