Resolve the flaky test ReplicatorTest.testCloseReplicatorStartProducer (

apache#1815)

### Motivation

Fixed the flaky behavior by ensuring the read operation completes with the expected failure instead of throwing assertion error in background.

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

@@ -796,7 +796,6 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
      */
     @Test(timeOut = 15000)
     public void testCloseReplicatorStartProducer() throws Exception {
-
         TopicName dest = TopicName.get("persistent://pulsar/ns1/closeCursor");
         // Producer on r1
         MessageProducer producer1 = new MessageProducer(url1, dest);
@@ -815,27 +814,20 @@ public void testCloseReplicatorStartProducer() throws Exception {
         ManagedCursor cursor = (ManagedCursor) cursorField.get(replicator);
         cursor.close();
         // try to read entries
-        CountDownLatch latch = new CountDownLatch(1);
         producer1.produce(10);
-        cursor.asyncReadEntriesOrWait(10, new ReadEntriesCallback() {
-            @Override
-            public void readEntriesComplete(List<Entry> entries, Object ctx) {
-                latch.countDown();
-                fail("it should have been failed");
-            }
 
-            @Override
-            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
-                latch.countDown();
-                assertTrue(exception instanceof CursorAlreadyClosedException);
-            }
-        }, null);
+        try {
+            cursor.readEntriesOrWait(10);
+            fail("It should have failed");
+        } catch (Exception e) {
+            assertEquals(e.getClass(), CursorAlreadyClosedException.class);
+        }
 
         // replicator-readException: cursorAlreadyClosed
         replicator.readEntriesFailed(new CursorAlreadyClosedException("Cursor already closed exception"), null);
 
         // wait replicator producer to be closed
-        Thread.sleep(1000);
+        Thread.sleep(100);
 
         // Replicator producer must be closed
         Field producerField = AbstractReplicator.class.getDeclaredField("producer");