Merge pull request #3 from merlimat/master

Fixed ZookeeperCacheTest failures on Travis build

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/main/java/com/yahoo/pulsar/broker/admin/Clusters.java

@@ -125,7 +125,9 @@ public class Clusters extends AdminResource {
         validatePoliciesReadOnlyAccess();
 
         try {
-            globalZk().setData(path("clusters", cluster), jsonMapper().writeValueAsBytes(clusterData), -1);
+            String clusterPath = path("clusters", cluster);
+            globalZk().setData(clusterPath, jsonMapper().writeValueAsBytes(clusterData), -1);
+            globalZkCache().invalidate(clusterPath);
             log.info("[{}] Updated cluster {}", clientAppId(), cluster);
         } catch (KeeperException.NoNodeException e) {
             log.warn("[{}] Failed to update cluster {}: Does not exist", clientAppId(), cluster);
@@ -186,8 +188,9 @@ public class Clusters extends AdminResource {
         }
 
         try {
-            globalZk().delete(path("clusters", cluster), -1);
-            clustersCache().invalidate(path("clusters", cluster));
+            String clusterPath = path("clusters", cluster);
+            globalZk().delete(clusterPath, -1);
+            globalZkCache().invalidate(clusterPath);
             log.info("[{}] Deleted cluster {}", clientAppId(), cluster);
         } catch (KeeperException.NoNodeException e) {
             log.warn("[{}] Failed to delete cluster {} - Does not exist", clientAppId(), cluster);

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/PersistentTopics.java

@@ -296,11 +296,12 @@ public class PersistentTopics extends AdminResource {
 
         try {
             // Write the new policies to zookeeper
-            globalZk().setData(path("policies", property, cluster, namespace), jsonMapper().writeValueAsBytes(policies),
-                    nodeStat.getVersion());
+            String namespacePath = path("policies", property, cluster, namespace);
+            globalZk().setData(namespacePath, jsonMapper().writeValueAsBytes(policies), nodeStat.getVersion());
 
             // invalidate the local cache to force update
-            policiesCache().invalidate(path("policies", property, cluster, namespace));
+            policiesCache().invalidate(namespacePath);
+            globalZkCache().invalidate(namespacePath);
 
             log.info("[{}] Successfully revoke access for role {} - destination {}", clientAppId(), role,
                     destinationUri);

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/Properties.java

@@ -150,7 +150,9 @@ public class Properties extends AdminResource {
                     throw new RestException(Status.CONFLICT, msg);
                 }
             }
-            globalZk().setData(path("policies", property), jsonMapper().writeValueAsBytes(newPropertyAdmin), -1);
+            String propertyPath = path("policies", property);
+            globalZk().setData(propertyPath, jsonMapper().writeValueAsBytes(newPropertyAdmin), -1);
+            globalZkCache().invalidate(propertyPath);
             log.info("[{}] updated property {}", clientAppId(), property);
         } catch (RestException re) {
             throw re;

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -279,7 +279,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
 
         // unloads all namespaces gracefully without disrupting mutually
         unloadNamespaceBundlesGracefully();
-        
+
         // close replication clients
         replicationClients.forEach((cluster, client) -> {
             try {
@@ -307,12 +307,14 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
      * <li>Second it starts unloading namespace bundle one by one without closing the connection in order to avoid
      * disruption for other namespacebundles which are sharing the same connection from the same client.</li>
      * <ul>
-     * 
+     *
      */
     public void unloadNamespaceBundlesGracefully() {
         try {
             // make broker-node unavailable from the cluster
-            pulsar.getLoadManager().disableBroker();
+            if (pulsar.getLoadManager() != null) {
+                pulsar.getLoadManager().disableBroker();
+            }
 
             // unload all namespace-bundles gracefully
             long closeTopicsStartTime = System.nanoTime();

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/ServerCnx.java

@@ -259,6 +259,7 @@ public class ServerCnx extends PulsarHandler {
                                 exception.getCause().getMessage()));
                     }
                     consumers.remove(consumerId, consumerFuture);
+
                     return null;
 
                 });
@@ -575,11 +576,11 @@ public class ServerCnx extends PulsarHandler {
         long producerId = producer.getProducerId();
         producers.remove(producerId);
         if(remoteEndpointProtocolVersion >= v5.getNumber()) {
-            ctx.writeAndFlush(Commands.newCloseProducer(producerId, -1L));    
+            ctx.writeAndFlush(Commands.newCloseProducer(producerId, -1L));
         } else {
             close();
         }
-        
+
     }
 
     public void closeConsumer(Consumer consumer) {
@@ -590,12 +591,12 @@ public class ServerCnx extends PulsarHandler {
         long consumerId = consumer.consumerId();
         consumers.remove(consumerId);
         if(remoteEndpointProtocolVersion >= v5.getNumber()) {
-            ctx.writeAndFlush(Commands.newCloseConsumer(consumerId, -1L));    
+            ctx.writeAndFlush(Commands.newCloseConsumer(consumerId, -1L));
         } else {
             close();
         }
     }
-    
+
     /**
      * It closes the connection with client which triggers {@code channelInactive()} which clears all producers and
      * consumers from connection-map
@@ -678,4 +679,8 @@ public class ServerCnx extends PulsarHandler {
     public String getRole() {
         return authRole;
     }
+
+    boolean hasConsumer(long consumerId) {
+        return consumers.containsKey(consumerId);
+    }
 }

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminApiTest.java

@@ -155,7 +155,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
     }
 
     @Test
-    public void clusters() throws PulsarAdminException {
+    public void clusters() throws Exception {
         admin.clusters().createCluster("usw",
                 new ClusterData("http://broker.messaging.use.example.com" + ":" + BROKER_WEBSERVICE_PORT));
         assertEquals(admin.clusters().getClusters(), Lists.newArrayList("use", "usw"));
@@ -178,6 +178,8 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
                         "https://new-broker.messaging.usw.example.com" + ":" + BROKER_WEBSERVICE_PORT_TLS));
 
         admin.clusters().deleteCluster("usw");
+        Thread.sleep(300);
+
         assertEquals(admin.clusters().getClusters(), Lists.newArrayList("use"));
 
         admin.namespaces().deleteNamespace("prop-xyz/use/ns1");

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminTest.java

@@ -30,12 +30,15 @@ import static org.testng.Assert.fail;
 
 import java.lang.reflect.Field;
 import java.net.URI;
+import java.security.Permissions;
+import java.security.acl.Permission;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 
 import javax.ws.rs.core.Response.Status;
 import javax.ws.rs.core.StreamingOutput;
@@ -605,6 +608,12 @@ public class AdminTest extends MockedPulsarServiceBaseTest {
         assertTrue(list.isEmpty());
         // create destination
         persistentTopics.createPartitionedTopic(property, cluster, namespace, destination, 5, false);
+
+        CountDownLatch notificationLatch = new CountDownLatch(2);
+        configurationCache.policiesCache().registerListener((path, data, stat) -> {
+            notificationLatch.countDown();
+        });
+
         // grant permission
         final Set<AuthAction> actions = Sets.newHashSet(AuthAction.produce);
         final String role = "test-role";
@@ -615,6 +624,10 @@ public class AdminTest extends MockedPulsarServiceBaseTest {
         assertEquals(permission.get(role), actions);
         // remove permission
         persistentTopics.revokePermissionsOnDestination(property, cluster, namespace, destination, role);
+
+        // Wait for cache to be updated
+        notificationLatch.await();
+
         // verify removed permission
         permission = persistentTopics.getPermissionsOnDestination(property, cluster, namespace, destination);
         assertTrue(permission.isEmpty());

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/PersistentTopicE2ETest.java

@@ -238,7 +238,7 @@ public class PersistentTopicE2ETest extends BrokerTestBase {
      * Validation: 1. validates active-cursor after active subscription 2. validate active-cursor with subscription 3.
      * unconsumed messages should be present into cache 4. cache and active-cursor should be empty once subscription is
      * closed
-     * 
+     *
      * @throws Exception
      */
     @Test
@@ -1033,13 +1033,14 @@ public class PersistentTopicE2ETest extends BrokerTestBase {
 
         Consumer consumer = pulsarClient.subscribe(topicName, "my-sub");
 
+        Message msg1 = MessageBuilder.create().setContent("message-1".getBytes()).build();
+        CompletableFuture<MessageId> future1 = producer.sendAsync(msg1);
+
         // Stop the broker, and publishes messages. Messages are accumulated in the producer queue and they're checksums
         // would have already been computed. If we change the message content at that point, it should result in a
         // checksum validation error
         stopBroker();
 
-        Message msg1 = MessageBuilder.create().setContent("message-1".getBytes()).build();
-        CompletableFuture<MessageId> future1 = producer.sendAsync(msg1);
 
         Message msg2 = MessageBuilder.create().setContent("message-2".getBytes()).build();
         CompletableFuture<MessageId> future2 = producer.sendAsync(msg2);

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

@@ -32,6 +32,7 @@ import static org.testng.Assert.assertTrue;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import javax.naming.AuthenticationException;
@@ -114,6 +115,9 @@ public class ServerCnxTest {
     private final String nonExistentTopicName = "persistent://nonexistent-prop/nonexistent-cluster/nonexistent-namespace/successNonExistentTopic";
     private final String topicWithNonLocalCluster = "persistent://prop/usw/ns-abc/successTopic";
 
+    private ManagedLedger ledgerMock = mock(ManagedLedger.class);
+    private ManagedCursor cursorMock = mock(ManagedCursor.class);
+
     @BeforeMethod
     public void setup() throws Exception {
         svcConfig = spy(new ServiceConfiguration());
@@ -611,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
@@ -632,6 +649,8 @@ public class ServerCnxTest {
                 producerName);
         channel.writeInbound(createProducer2);
 
+        successTopicCreationDelayLatch.countDown();
+
         // Close succeeds
         Object response = getResponse();
         assertEquals(response.getClass(), CommandSuccess.class);
@@ -649,11 +668,24 @@ public class ServerCnxTest {
         channel.finish();
     }
 
-    @Test(timeOut = 30000)
+    @Test(timeOut = 30000, enabled = false)
     public void testCreateProducerMultipleTimeouts() throws Exception {
         resetChannel();
         setChannelConnected();
 
+        // Delay the topic creation in a deterministic way
+        CountDownLatch topicCreationDelayLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                topicCreationDelayLatch.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
@@ -688,6 +720,9 @@ public class ServerCnxTest {
         assertEquals(response.getClass(), CommandSuccess.class);
         assertEquals(((CommandSuccess) response).getRequestId(), 2);
 
+        // Now allow topic creation to complete
+        topicCreationDelayLatch.countDown();
+
         // 1st producer it's not acked
 
         // 2nd producer fails
@@ -719,6 +754,31 @@ public class ServerCnxTest {
         resetChannel();
         setChannelConnected();
 
+        // Delay the topic creation in a deterministic way
+        CountDownLatch failedTopicCreationDelayLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                failedTopicCreationDelayLatch.await();
+
+                ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null);
+                return null;
+            }
+        }).when(mlFactoryMock).asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class),
+                any(OpenLedgerCallback.class), anyObject());
+
+        CountDownLatch topicCreationDelayLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                topicCreationDelayLatch.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 a failure producer which will timeout creation after 100msec
         // 2. close producer
@@ -741,6 +801,9 @@ public class ServerCnxTest {
                 producerName);
         channel.writeInbound(createProducer2);
 
+        failedTopicCreationDelayLatch.countDown();
+        topicCreationDelayLatch.countDown();
+
         // Close succeeds
         Object response = getResponse();
         assertEquals(response.getClass(), CommandSuccess.class);
@@ -752,7 +815,7 @@ public class ServerCnxTest {
         assertEquals(((CommandError) response).getRequestId(), 3);
 
         // Wait till the failtopic timeout interval
-        Thread.sleep(200);
+        Thread.sleep(500);
         ByteBuf createProducer3 = Commands.newProducer(successTopicName, 1 /* producer id */, 4 /* request id */,
                 producerName);
         channel.writeInbound(createProducer3);
@@ -762,7 +825,7 @@ public class ServerCnxTest {
         assertEquals(response.getClass(), CommandProducerSuccess.class);
         assertEquals(((CommandProducerSuccess) response).getRequestId(), 4);
 
-        Thread.sleep(100);
+        Thread.sleep(500);
 
         // We should not receive response for 1st producer, since it was cancelled by the close
         assertTrue(channel.outboundMessages().isEmpty());
@@ -776,6 +839,21 @@ public class ServerCnxTest {
         resetChannel();
         setChannelConnected();
 
+        // Delay the topic creation in a deterministic way
+        CountDownLatch topicCreationDelayLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                topicCreationDelayLatch.await();
+
+                synchronized (ServerCnxTest.this) {
+                    ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null);
+                }
+                return null;
+            }
+        }).when(mlFactoryMock).asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class),
+                any(OpenLedgerCallback.class), anyObject());
+
         // In a subscribe timeout from client side we expect to see this sequence of commands :
         // 1. Subscribe
         // 2. close consumer (when the timeout is triggered, which may be before the consumer was created on the broker)
@@ -803,31 +881,33 @@ public class ServerCnxTest {
                 successSubName, 1 /* consumer id */, 5 /* request id */, SubType.Exclusive, "test" /* consumer name */);
         channel.writeInbound(subscribe4);
 
-        Object response;
+        topicCreationDelayLatch.countDown();
 
-        // Close succeeds
-        response = getResponse();
-        assertEquals(response.getClass(), CommandSuccess.class);
-        assertEquals(((CommandSuccess) response).getRequestId(), 2);
+        Object response;
 
-        // All other subscribe should fail
-        response = getResponse();
-        assertEquals(response.getClass(), CommandError.class);
-        assertEquals(((CommandError) response).getRequestId(), 3);
+        synchronized (this) {
+            // Close succeeds
+            response = getResponse();
+            assertEquals(response.getClass(), CommandSuccess.class);
+            assertEquals(((CommandSuccess) response).getRequestId(), 2);
 
-        response = getResponse();
-        assertEquals(response.getClass(), CommandError.class);
-        assertEquals(((CommandError) response).getRequestId(), 4);
+            // All other subscribe should fail
+            response = getResponse();
+            assertEquals(response.getClass(), CommandError.class);
+            assertEquals(((CommandError) response).getRequestId(), 3);
 
-        response = getResponse();
-        assertEquals(response.getClass(), CommandError.class);
-        assertEquals(((CommandError) response).getRequestId(), 5);
+            response = getResponse();
+            assertEquals(response.getClass(), CommandError.class);
+            assertEquals(((CommandError) response).getRequestId(), 4);
 
-        Thread.sleep(100);
+            response = getResponse();
+            assertEquals(response.getClass(), CommandError.class);
+            assertEquals(((CommandError) response).getRequestId(), 5);
 
-        // We should not receive response for 1st producer, since it was cancelled by the close
-        assertTrue(channel.outboundMessages().isEmpty());
-        assertTrue(channel.isActive());
+            // We should not receive response for 1st producer, since it was cancelled by the close
+            assertTrue(channel.outboundMessages().isEmpty());
+            assertTrue(channel.isActive());
+        }
 
         channel.finish();
     }
@@ -837,6 +917,39 @@ 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());
+
+        CountDownLatch failedTopicCreationDelayLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+
+                new Thread(() -> {
+                    try {
+                        failedTopicCreationDelayLatch.await();
+                    } catch (InterruptedException e) {
+                    }
+
+                    ((OpenLedgerCallback) invocationOnMock.getArguments()[2])
+                            .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null);
+                }).start();
+
+                return null;
+            }
+        }).when(mlFactoryMock).asyncOpen(matches(".*fail.*"), any(ManagedLedgerConfig.class),
+                any(OpenLedgerCallback.class), anyObject());
+
         // In a subscribe timeout from client side we expect to see this sequence of commands :
         // 1. Subscribe against failtopic which will fail after 100msec
         // 2. close consumer
@@ -845,7 +958,6 @@ public class ServerCnxTest {
 
         // These operations need to be serialized, to allow the last subscribe operation to finally succeed
         // (There can be more subscribe/close pairs in the sequence, depending on the client timeout
-
         ByteBuf subscribe1 = Commands.newSubscribe(failTopicName, //
                 successSubName, 1 /* consumer id */, 1 /* request id */, SubType.Exclusive, "test" /* consumer name */);
         channel.writeInbound(subscribe1);
@@ -857,6 +969,9 @@ public class ServerCnxTest {
                 successSubName, 1 /* consumer id */, 3 /* request id */, SubType.Exclusive, "test" /* consumer name */);
         channel.writeInbound(subscribe2);
 
+        successTopicCreationDelayLatch.countDown();
+        failedTopicCreationDelayLatch.countDown();
+
         Object response;
 
         // Close succeeds
@@ -869,8 +984,9 @@ public class ServerCnxTest {
         assertEquals(response.getClass(), CommandError.class);
         assertEquals(((CommandError) response).getRequestId(), 3);
 
-        // Wait till the failtopic timeout interval
-        Thread.sleep(200);
+        while (serverCnx.hasConsumer(1)) {
+            Thread.sleep(10);
+        }
 
         ByteBuf subscribe3 = Commands.newSubscribe(successTopicName, //
                 successSubName, 1 /* consumer id */, 4 /* request id */, SubType.Exclusive, "test" /* consumer name */);
@@ -1047,16 +1163,13 @@ public class ServerCnxTest {
     }
 
     private void setupMLAsyncCallbackMocks() {
-        final ManagedLedger ledgerMock = mock(ManagedLedger.class);
-        final ManagedCursor cursorMock = mock(ManagedCursor.class);
-
         doReturn(new ArrayList<Object>()).when(ledgerMock).getCursors();
 
         // call openLedgerComplete with ledgerMock on ML factory asyncOpen
         doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
-                Thread.sleep(100);
+                Thread.sleep(300);
                 ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null);
                 return null;
             }
@@ -1067,7 +1180,7 @@ public class ServerCnxTest {
         doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
-                Thread.sleep(100);
+                Thread.sleep(300);
                 new Thread(() -> {
                     ((OpenLedgerCallback) invocationOnMock.getArguments()[2])
                             .openLedgerFailed(new ManagedLedgerException("Managed ledger failure"), null);
@@ -1091,6 +1204,7 @@ public class ServerCnxTest {
         doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                Thread.sleep(300);
                 ((OpenCursorCallback) invocationOnMock.getArguments()[1]).openCursorComplete(cursorMock, null);
                 return null;
             }
@@ -1099,6 +1213,7 @@ public class ServerCnxTest {
         doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+                Thread.sleep(300);
                 ((OpenCursorCallback) invocationOnMock.getArguments()[1])
                         .openCursorFailed(new ManagedLedgerException("Managed ledger failure"), null);
                 return null;

pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/ClientErrorsTest.java

@@ -39,6 +39,7 @@ import com.yahoo.pulsar.client.api.ConsumerConfiguration;
 import com.yahoo.pulsar.client.api.Producer;
 import com.yahoo.pulsar.client.api.PulsarClient;
 import com.yahoo.pulsar.client.api.PulsarClientException;
+import com.yahoo.pulsar.client.api.PulsarClientException.LookupException;
 import com.yahoo.pulsar.client.api.SubscriptionType;
 import com.yahoo.pulsar.client.impl.ConsumerBase;
 import com.yahoo.pulsar.client.impl.ProducerBase;
@@ -374,7 +375,7 @@ public class ClientErrorsTest {
 
     private void subscribeFailAfterRetryTimeout(String topic) throws Exception {
         ClientConfiguration conf = new ClientConfiguration();
-        conf.setOperationTimeout(1, TimeUnit.SECONDS);
+        conf.setOperationTimeout(200, TimeUnit.MILLISECONDS);
 
         PulsarClient client = PulsarClient.create("http://127.0.0.1:" + WEB_SERVICE_PORT, conf);
         final AtomicInteger counter = new AtomicInteger(0);
@@ -382,7 +383,7 @@ public class ClientErrorsTest {
         mockBrokerService.setHandleSubscribe((ctx, subscribe) -> {
             if (counter.incrementAndGet() == 2) {
                 try {
-                    Thread.sleep(2000);
+                    Thread.sleep(500);
                 } catch (InterruptedException e) {
                     // do nothing
                 }
@@ -393,11 +394,11 @@ public class ClientErrorsTest {
         try {
             ConsumerConfiguration cConf = new ConsumerConfiguration();
             cConf.setSubscriptionType(SubscriptionType.Exclusive);
-            Consumer consumer = client.subscribe(topic, "sub1", cConf);
+            client.subscribe(topic, "sub1", cConf);
             fail("Should have failed");
         } catch (Exception e) {
             // we fail even on the retriable error
-            assertTrue(e instanceof PulsarClientException.LookupException);
+            assertEquals(e.getClass(), LookupException.class);
         }
 
         mockBrokerService.resetHandleSubscribe();
@@ -576,6 +577,7 @@ public class ClientErrorsTest {
 
         // close the cnx after creating the producer
         channelCtx.get().channel().close();
+        Thread.sleep(300);
 
         producer.send(new byte[0]);
 

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
@@ -277,8 +279,6 @@ public class ZookeeperCacheTest {
         // seen by the cache
         zkClient.failAfter(-1, Code.OK);
         zkClient.delete("/my_test2", -1);
-        // Make sure it has not been updated yet
-        assertEquals(zkCache.get("/my_test2"), value);
         zkCacheService.process(new WatchedEvent(Event.EventType.None, KeeperState.SyncConnected, null));
         assertEquals(zkCache.get("/other"), newValue);