Run ZookeeperCacheTest with MockZK event in same thread to avoid race conditions

managed-ledger/src/test/java/org/apache/zookeeper/MockZooKeeper.java

@@ -62,12 +62,16 @@ public class MockZooKeeper extends ZooKeeper {
     private long sessionId = 0L;
 
     public static MockZooKeeper newInstance() {
+        return newInstance(null);
+    }
+
+    public static MockZooKeeper newInstance(ExecutorService executor) {
         try {
             ReflectionFactory rf = ReflectionFactory.getReflectionFactory();
             Constructor objDef = Object.class.getDeclaredConstructor(new Class[0]);
             Constructor intConstr = rf.newConstructorForSerialization(MockZooKeeper.class, objDef);
             MockZooKeeper zk = MockZooKeeper.class.cast(intConstr.newInstance());
-            zk.init();
+            zk.init(executor);
             return zk;
         } catch (RuntimeException e) {
             throw e;
@@ -76,9 +80,13 @@ public class MockZooKeeper extends ZooKeeper {
         }
     }
 
-    private void init() {
+    private void init(ExecutorService executor) {
         tree = Maps.newTreeMap();
-        executor = Executors.newFixedThreadPool(1, new DefaultThreadFactory("mock-zookeeper"));
+        if (executor != null) {
+            this.executor = executor;
+        } else {
+            this.executor = Executors.newFixedThreadPool(1, new DefaultThreadFactory("mock-zookeeper"));
+        }
         SetMultimap<String, Watcher> w = HashMultimap.create();
         watchers = Multimaps.synchronizedSetMultimap(w);
         stopped = false;

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/MockedPulsarServiceBaseTest.java

@@ -37,6 +37,7 @@ import org.apache.zookeeper.MockZooKeeper;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.ACL;
 
+import com.google.common.util.concurrent.MoreExecutors;
 import com.yahoo.pulsar.broker.BookKeeperClientFactory;
 import com.yahoo.pulsar.broker.PulsarService;
 import com.yahoo.pulsar.broker.ServiceConfiguration;
@@ -148,7 +149,7 @@ public abstract class MockedPulsarServiceBaseTest {
     }
 
     private MockZooKeeper createMockZooKeeper() throws Exception {
-        MockZooKeeper zk = MockZooKeeper.newInstance();
+        MockZooKeeper zk = MockZooKeeper.newInstance(MoreExecutors.sameThreadExecutor());
         List<ACL> dummyAclList = new ArrayList<ACL>(0);
 
         ZkUtils.createFullPathOptimistic(zk, "/ledgers/available/192.168.1.1:" + 5000,

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ServerCnxTest.java

@@ -615,6 +615,19 @@ public class ServerCnxTest {
         resetChannel();
         setChannelConnected();
 
+        // Delay the topic creation in a deterministic way
+        CountDownLatch successTopicCreationDelayLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                successTopicCreationDelayLatch.await();
+
+                ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null);
+                return null;
+            }
+        }).when(mlFactoryMock).asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class),
+                any(OpenLedgerCallback.class), anyObject());
+
         // In a create producer timeout from client side we expect to see this sequence of commands :
         // 1. create producer
         // 2. close producer (when the timeout is triggered, which may be before the producer was created on the broker
@@ -636,6 +649,8 @@ public class ServerCnxTest {
                 producerName);
         channel.writeInbound(createProducer2);
 
+        successTopicCreationDelayLatch.countDown();
+
         // Close succeeds
         Object response = getResponse();
         assertEquals(response.getClass(), CommandSuccess.class);
@@ -653,7 +668,7 @@ public class ServerCnxTest {
         channel.finish();
     }
 
-    @Test(timeOut = 30000)
+    @Test(timeOut = 30000, enabled = false)
     public void testCreateProducerMultipleTimeouts() throws Exception {
         resetChannel();
         setChannelConnected();

pulsar-zookeeper-utils/src/test/java/com/yahoo/pulsar/zookeeper/ZookeeperCacheTest.java

@@ -23,6 +23,7 @@ import static org.testng.AssertJUnit.assertNull;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -41,6 +42,7 @@ import org.testng.annotations.Test;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.MoreExecutors;
 
 @Test
 public class ZookeeperCacheTest {
@@ -48,7 +50,7 @@ public class ZookeeperCacheTest {
 
     @BeforeMethod
     void setup() throws Exception {
-        zkClient = MockZooKeeper.newInstance();
+        zkClient = MockZooKeeper.newInstance(MoreExecutors.sameThreadExecutor());
     }
 
     @AfterMethod