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 static MockZooKeeper newInstance() {
         }
     }
 
-    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 void updateCluster(@PathParam("cluster") String cluster, ClusterData clus
         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 void deleteCluster(@PathParam("cluster") String cluster) {
         }
 
         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 void revokePermissionsOnDestination(@PathParam("property") String propert
 
         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 void updateProperty(@PathParam("property") String property, PropertyAdmin
                     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 void close() throws IOException {
 
         // unloads all namespaces gracefully without disrupting mutually
         unloadNamespaceBundlesGracefully();
-        
+
         // close replication clients
         replicationClients.forEach((cluster, client) -> {
             try {
@@ -307,12 +307,14 @@ public void close() throws IOException {
      * <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 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                 exception.getCause().getMessage()));
                     }
                     consumers.remove(consumerId, consumerFuture);
+
                     return null;
 
                 });
@@ -575,11 +576,11 @@ public void closeProducer(Producer producer) {
         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 void closeConsumer(Consumer consumer) {
         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 BrokerService getBrokerService() {
     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 static Object[][] bundling() {
     }
 
     @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 void clusters() throws PulsarAdminException {
                         "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 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 @@ void persistentTopics() throws Exception {
         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 @@ void persistentTopics() throws Exception {
         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.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 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception {
     }
 
     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 void testConsumerFlowControl() throws Exception {
      * 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 void testPayloadCorruptionDetection() throws Exception {
 
         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);