[fix][test] Fix flaky NonPersistentTopicTest.testMsgDropStat (#24134)

Author: pdolif

pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java

@@ -871,14 +871,15 @@ public void testMsgDropStat() throws Exception {
                 .messageRoutingMode(MessageRoutingMode.SinglePartition)
                 .create();
             @Cleanup("shutdownNow")
-            ExecutorService executor = Executors.newFixedThreadPool(5);
+            ExecutorService executor = Executors.newFixedThreadPool(10);
             byte[] msgData = "testData".getBytes();
             final int totalProduceMessages = 1000;
             CountDownLatch latch = new CountDownLatch(1);
             AtomicInteger messagesSent = new AtomicInteger(0);
             for (int i = 0; i < totalProduceMessages; i++) {
                 executor.submit(() -> {
-                    producer.sendAsync(msgData).handle((msgId, e) -> {
+                    try {
+                        MessageId msgId = producer.send(msgData);
                         int count = messagesSent.incrementAndGet();
                         // process at least 20% of messages before signalling the latch
                         // a non-persistent message will return entryId as -1 when it has been dropped
@@ -888,8 +889,14 @@ public void testMsgDropStat() throws Exception {
                                 && ((MessageIdImpl) msgId).getEntryId() == -1) {
                             latch.countDown();
                         }
-                        return null;
-                    });
+
+                        Thread.sleep(10);
+                    } catch (PulsarClientException e) {
+                        throw new RuntimeException(e);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        throw new RuntimeException(e);
+                    }
                 });
             }
             assertTrue(latch.await(5, TimeUnit.SECONDS));