Async ZooKeeper cache implementation #76

merlimat · 2016-10-23T22:26:17Z

Motivation

Current implementation of Pulsar ZooKeeper cache is synchronous and based on Guava
cache class. Whenever a get() request doesn't find an object in the cache, it will
be fetched from ZooKeeper, deserialized and inserted into cache.

The synchronous nature of Guava cache, forces us to use sync ZK API, since the
cache loading has to be performed atomically.

To enable ZK cache lookups from IO threads, we need to be able to change
cache interface to expose asynchronous methods, that returns a CompletableFuture
rather than the real object.

This change is needed by #5

Modifications

Changed Guava cache with Caffeine async loading cache
Added async lookup methods in ZooKeeperDataCache
Changed namespace bundle ownership registration to be asynchronous
Use Optional<> to treat ZK NoNodeError instead of propagating the exception to the cache
Changed HTTP topic lookup to use async servlet
Fixed MockZookeeper to notify watchers without holding instance mutex to avoid deadlocks in unit tests

Result

All cache operations async
Callers of ZK cache will need to explicitly deal with the z-node not being there
HTTP lookups handled more efficiently (less number of threads required)

yahoocla · 2016-10-23T22:26:20Z

CLA is valid!

yahoocla · 2016-10-23T22:26:20Z

CLA is valid!

yahoocla · 2016-10-23T22:26:20Z

CLA is valid!

maskit

Needs few small fixes but the change seems reasonable.

maskit · 2016-10-24T20:57:55Z

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/SLAMonitoringTest.java

@@ -202,6 +204,8 @@ public void testUnloadIfBrokerCrashes() {

        log.info("Lookup for namespace {}", destination);

+        System.err.println("CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC");


Should remove or replace with a meaningful message.

Ups, I forgot to remove these debugging lines

maskit · 2016-10-24T21:13:32Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceService.java

            policies = this.pulsar.getLocalZkCacheService().policiesCache().get(path);
        }
-        policies.bundles = getBundlesData(nsBundles);
+
+        policies.get().bundles = getBundlesData(nsBundles);


I'm not 100% sure but policies.get() can be null here if creating new policies failed.

If creating fails, it should throw an exception and thus skip that line.

maskit · 2016-10-24T21:27:08Z

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

                        // TODO: make this admin call asynchronous
                        pulsar().getAdminClient().namespaces().unsubscribeNamespaceBundle(nsName.toString(),
                                nsBundle.getBundleRange(), subscription);
                    }
                } catch (Exception e) {
+                    e.printStackTrace();


Is this intended or just left behind?

Good catch, I forgot it here

rdhabalia · 2016-10-24T21:44:49Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCache.java

@@ -212,34 +222,60 @@ public NamespaceEphemeralData getOwner(ServiceUnitId suname) throws Exception {
     * @return The ephemeral node data showing the current ownership info in <code>ZooKeeper</code>


Minor: should we replace ServiceUnit with NamespaceBundle at method-docs.

rdhabalia · 2016-10-24T22:58:44Z

pulsar-zookeeper-utils/src/main/java/com/yahoo/pulsar/zookeeper/ZooKeeperCache.java

+                if (rc == Code.OK.intValue()) {
+                    try {
+                        T obj = deserializer.deserialize(path, content);
+                        zkFuture.complete(new AbstractMap.SimpleEntry<Object, Stat>(obj, stat));


can we create AbstractMap.SimpleImmutableEntry here.?

rdhabalia · 2016-10-24T23:00:05Z

pulsar-zookeeper-utils/src/main/java/com/yahoo/pulsar/zookeeper/ZooKeeperCache.java

+            return zkFuture;
+        }).thenAccept(result -> {
+            if (result != null) {
+                future.complete(Optional


Instead creating a new object can we just cast the result:
future.complete(Optional.of(((Entry<T, Stat>) result)));

rdhabalia · 2016-10-24T23:07:38Z

pulsar-broker/src/main/java/com/yahoo/pulsar/utils/CopyOnWriteArrayList.java

@@ -119,4 +120,17 @@ public void replaceAll(UnaryOperator operator) {
        return EMPTY_LIST;
    }

+    public static void main(String[] args) {


should we move it to test class.

rdhabalia · 2016-10-25T20:35:20Z

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

        } catch (Exception e) {
            log.error("[{}] Failed to get cluster {}", clientAppId(), cluster, e);
-            throw new RestException(e);
+            if (e instanceof RestException) {


Is there any usecase which can throw exception other than RestException? If not then we may not need this try-catch block.

Yes, we could get ZK error when trying to read

rdhabalia · 2016-10-25T20:41:21Z

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

-                    .get(path("clusters", cluster, "namespaceIsolationPolicies"));
+                    .get(path("clusters", cluster, "namespaceIsolationPolicies"))
+                    .orElseThrow(() -> new RestException(Status.NOT_FOUND,
+                            "NamespaceIsolationPolicies for cluster " + cluster + " does not exist"));


should we log warn-message where-ever we are throwing RestException in ``orElseThrow`, which may help at broker-debugging incase if requires.

Yes, the error is being logged at line 227

rdhabalia · 2016-10-25T20:42:35Z

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

    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
            @ApiResponse(code = 404, message = "Cluster doesn't exist") })
    public Map<String, NamespaceIsolationData> getNamespaceIsolationPolicies(@PathParam("cluster") String cluster)
            throws Exception {
        validateSuperUserAccess();
-        validateClusterExists(cluster);
+        if (!clustersCache().get(path("clusters", cluster)).isPresent()) {


can't we use validateClusterExists() method?

validateClusterExists() throws 412 HTTP error if the cluster doesn't exist. But in this case we to return 404.

rdhabalia · 2016-10-25T20:58:42Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/cache/LocalZooKeeperCacheService.java

-    public void createPolicies(String path, boolean readFromGlobal) throws Exception {
+    @SuppressWarnings("deprecation")
+    public CompletableFuture<Optional<LocalPolicies>> createPolicies(String path, boolean readFromGlobal) {
+        checkNotNull(path, "path can't be null");


instead throwing runtime-exception, should we just complete future with failure. As caller of getAsync() might not be handling the exception ie: OwnershipCache.getOwnerAsync()

In this case is more of a sanity check.. we could even remove the check alltogether

rdhabalia · 2016-10-25T21:05:35Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/cache/LocalZooKeeperCacheService.java

+                    if (LOG.isDebugEnabled()) {
+                        LOG.debug("Global policies not found at {}", globalPath);
+                    }
+                    future.complete(Optional.empty());


should it readFromGlobalFuture.complete(Optional.empty()); or we can return future as it's already complete.

Good catch! I guess both options would work here.

rdhabalia · 2016-10-25T23:51:14Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceService.java

-        }
-        checkNotNull(serviceUnit);
-
+    public boolean isServiceUnitDisabled(NamespaceBundle bundle) throws Exception {


Minor: as this method is under NamespaceService and we have changed argument as NamespaceBundle then should we rename this method as : isNamespaceBundleDisabled()

rdhabalia · 2016-10-26T00:06:24Z

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

@@ -151,6 +153,14 @@ public BrokerService(PulsarService pulsar) throws Exception {
        this.pulsarStats = new PulsarStats(pulsar);
        this.offlineTopicStatCache = new ConcurrentOpenHashMap<>();

+        ServiceConfiguration serviceConfig = pulsar.getConfiguration();
+        this.defaultPolicies = new Policies();


Not sure if we use defaultPolicies anywhere.?

Good point. That initialization was moved to when the topic is loaded.

rdhabalia · 2016-10-26T00:32:54Z

pulsar-zookeeper-utils/src/main/java/com/yahoo/pulsar/zookeeper/ZooKeeperDataCache.java

+    public CompletableFuture<Optional<T>> getAsync(String path) {
+        CompletableFuture<Optional<T>> future = new CompletableFuture<>();
+        cache.getDataAsync(path, this, this).thenAccept(entry -> {
+            future.complete(entry.map(Entry::getKey));


I think there is a change of NPE:
ZookeeperCache.getDataAsync returns null as future-result in case of NoNodeException

Actually, this should be OK, the zkFuture.complete(null) is to make sure the AsyncLoadingCache will not retain the entry in the map. That null value is then converted into an Optional.empty() :

}).thenAccept(result -> { if (result != null) { future.complete(Optional.of((Entry<T, Stat>) result)); } else { future.complete(Optional.empty()); }

yes.. I missed the last part of ZookeeperCache.getDataAsync. We should be good.

rdhabalia

Looks good to me. 👍

* Name our threads * We already set the name of runner thread

merlimat added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label Oct 23, 2016

merlimat added this to the 1.16 milestone Oct 23, 2016

merlimat self-assigned this Oct 23, 2016

merlimat force-pushed the async-zk-cache branch 2 times, most recently from d9fda91 to 2d88225 Compare October 24, 2016 01:36

maskit requested changes Oct 24, 2016

View reviewed changes

rdhabalia reviewed Oct 24, 2016

View reviewed changes

merlimat force-pushed the async-zk-cache branch from 2d88225 to 126e600 Compare October 25, 2016 04:14

merlimat added 5 commits October 25, 2016 10:23

Async ZK cache load

4c7cc7f

Using Optional<> for no-node-exception

c1191a0

Test fixes

c2aa81b

Addressed PR comments

1cb83d5

Corrected class names in Javadoc

238faf8

merlimat force-pushed the async-zk-cache branch from 88154aa to 238faf8 Compare October 25, 2016 17:23

rdhabalia reviewed Oct 25, 2016

View reviewed changes

Fixed readFromGlobal zk read future

5b29a85

rdhabalia reviewed Oct 26, 2016

View reviewed changes

More minor fixes

3923316

rdhabalia approved these changes Oct 26, 2016

View reviewed changes

merlimat merged commit 1868d2f into apache:master Oct 26, 2016

merlimat mentioned this pull request Oct 26, 2016

Support Topic lookup using Pulsar binary protocol #5

Merged

merlimat deleted the async-zk-cache branch October 28, 2016 19:13

sijie pushed a commit to sijie/pulsar that referenced this pull request Mar 4, 2018

Name our threads (apache#76)

992e947

* Name our threads * We already set the name of runner thread

xiaotongwang1 mentioned this pull request Aug 4, 2021

Pulsar 2.7.0+ KOP 2.7.2.x getPartitionedTopicMetadata timeout #11532

Closed

ZHr-UChiHa mentioned this pull request Jun 3, 2023

[Bug] python pulsar client create producer thread suspend because of deadlock apache/pulsar-client-python#129

Open

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Async ZooKeeper cache implementation #76

Async ZooKeeper cache implementation #76

merlimat commented Oct 23, 2016 •

edited

Loading

yahoocla commented Oct 23, 2016

yahoocla commented Oct 23, 2016

yahoocla commented Oct 23, 2016

maskit left a comment

maskit Oct 24, 2016

merlimat Oct 24, 2016

maskit Oct 24, 2016

merlimat Oct 24, 2016

maskit Oct 24, 2016

merlimat Oct 24, 2016

rdhabalia Oct 24, 2016

merlimat Oct 25, 2016

rdhabalia Oct 24, 2016

rdhabalia Oct 24, 2016

rdhabalia Oct 24, 2016

rdhabalia Oct 25, 2016

merlimat Oct 25, 2016

rdhabalia Oct 25, 2016

merlimat Oct 25, 2016

rdhabalia Oct 25, 2016

merlimat Oct 25, 2016

rdhabalia Oct 25, 2016

merlimat Oct 25, 2016

rdhabalia Oct 25, 2016

merlimat Oct 25, 2016

rdhabalia Oct 25, 2016

rdhabalia Oct 26, 2016

merlimat Oct 26, 2016

rdhabalia Oct 26, 2016

merlimat Oct 26, 2016

rdhabalia Oct 26, 2016

rdhabalia left a comment

		@@ -202,6 +204,8 @@ public void testUnloadIfBrokerCrashes() {

		log.info("Lookup for namespace {}", destination);

		System.err.println("CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC");

		@@ -212,34 +222,60 @@ public NamespaceEphemeralData getOwner(ServiceUnitId suname) throws Exception {
		* @return The ephemeral node data showing the current ownership info in <code>ZooKeeper</code>

Async ZooKeeper cache implementation #76

Async ZooKeeper cache implementation #76

Conversation

merlimat commented Oct 23, 2016 • edited Loading

Motivation

Modifications

Result

yahoocla commented Oct 23, 2016

yahoocla commented Oct 23, 2016

yahoocla commented Oct 23, 2016

maskit left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia left a comment

Choose a reason for hiding this comment

merlimat commented Oct 23, 2016 •

edited

Loading