distributedlog-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [05/29] incubator-distributedlog git commit: DL: remove watches when unregister children watches
Date Wed, 21 Dec 2016 08:00:13 GMT
DL: remove watches when unregister children watches

RB_ID=833858


Project: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/commit/b571d3b4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/tree/b571d3b4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/diff/b571d3b4

Branch: refs/heads/merge/DL-98
Commit: b571d3b4adcc140acca881979782474c27459d34
Parents: 517c77c
Author: Sijie Guo <sijieg@twitter.com>
Authored: Mon May 23 21:01:57 2016 -0700
Committer: Sijie Guo <sijieg@twitter.com>
Committed: Mon Dec 12 16:35:26 2016 -0800

----------------------------------------------------------------------
 .../twitter/distributedlog/ZooKeeperClient.java |  1 +
 .../distributedlog/zk/ZKWatcherManager.java     | 34 +++++++++++++++++++-
 .../distributedlog/zk/TestZKWatcherManager.java |  1 +
 3 files changed, 35 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b571d3b4/distributedlog-core/src/main/java/com/twitter/distributedlog/ZooKeeperClient.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/ZooKeeperClient.java
b/distributedlog-core/src/main/java/com/twitter/distributedlog/ZooKeeperClient.java
index 912d592..9ea9e37 100644
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/ZooKeeperClient.java
+++ b/distributedlog-core/src/main/java/com/twitter/distributedlog/ZooKeeperClient.java
@@ -169,6 +169,7 @@ public class ZooKeeperClient {
         this.credentials = credentials;
         this.watcherManager = ZKWatcherManager.newBuilder()
                 .name(name)
+                .zkc(this)
                 .statsLogger(statsLogger.scope("watcher_manager"))
                 .build();
     }

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b571d3b4/distributedlog-core/src/main/java/com/twitter/distributedlog/zk/ZKWatcherManager.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/zk/ZKWatcherManager.java
b/distributedlog-core/src/main/java/com/twitter/distributedlog/zk/ZKWatcherManager.java
index 4068737..a24b560 100644
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/zk/ZKWatcherManager.java
+++ b/distributedlog-core/src/main/java/com/twitter/distributedlog/zk/ZKWatcherManager.java
@@ -17,8 +17,11 @@
  */
 package com.twitter.distributedlog.zk;
 
+import com.twitter.distributedlog.ZooKeeperClient;
 import org.apache.bookkeeper.stats.Gauge;
 import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.slf4j.Logger;
@@ -50,31 +53,40 @@ public class ZKWatcherManager implements Watcher {
 
         private String _name;
         private StatsLogger _statsLogger;
+        private ZooKeeperClient _zkc;
 
         public Builder name(String name) {
             this._name = name;
             return this;
         }
 
+        public Builder zkc(ZooKeeperClient zkc) {
+            this._zkc = zkc;
+            return this;
+        }
+
         public Builder statsLogger(StatsLogger statsLogger) {
             this._statsLogger = statsLogger;
             return this;
         }
 
         public ZKWatcherManager build() {
-            return new ZKWatcherManager(_name, _statsLogger);
+            return new ZKWatcherManager(_name, _zkc, _statsLogger);
         }
     }
 
     private final String name;
+    private final ZooKeeperClient zkc;
     private final StatsLogger statsLogger;
 
     protected final ConcurrentMap<String, Set<Watcher>> childWatches;
     protected final AtomicInteger allWatchesGauge;
 
     private ZKWatcherManager(String name,
+                             ZooKeeperClient zkc,
                              StatsLogger statsLogger) {
         this.name = name;
+        this.zkc = zkc;
         this.statsLogger = statsLogger;
 
         // watches
@@ -141,6 +153,26 @@ public class ZKWatcherManager implements Watcher {
                 logger.warn("Remove a non-registered child watcher {} from path {}", watcher,
path);
             }
             if (watchers.isEmpty()) {
+                // best-efforts to remove watches
+                try {
+                    if (null != zkc) {
+                        zkc.get().removeWatches(path, this, WatcherType.Children, true, new
AsyncCallback.VoidCallback() {
+                            @Override
+                            public void processResult(int rc, String path, Object ctx) {
+                                if (KeeperException.Code.OK.intValue() == rc) {
+                                    logger.debug("Successfully removed children watches from
{}", path);
+                                } else {
+                                    logger.debug("Encountered exception on removing children
watches from {}",
+                                            path, KeeperException.create(KeeperException.Code.get(rc)));
+                                }
+                            }
+                        }, null);
+                    }
+                } catch (InterruptedException e) {
+                    logger.debug("Encountered exception on removing watches from {}", path,
e);
+                } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
+                    logger.debug("Encountered exception on removing watches from {}", path,
e);
+                }
                 childWatches.remove(path, watchers);
             }
         }

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b571d3b4/distributedlog-core/src/test/java/com/twitter/distributedlog/zk/TestZKWatcherManager.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/test/java/com/twitter/distributedlog/zk/TestZKWatcherManager.java
b/distributedlog-core/src/test/java/com/twitter/distributedlog/zk/TestZKWatcherManager.java
index ee00ab9..6f269c3 100644
--- a/distributedlog-core/src/test/java/com/twitter/distributedlog/zk/TestZKWatcherManager.java
+++ b/distributedlog-core/src/test/java/com/twitter/distributedlog/zk/TestZKWatcherManager.java
@@ -34,6 +34,7 @@ public class TestZKWatcherManager {
     public void testRegisterUnregisterWatcher() throws Exception {
         ZKWatcherManager watcherManager = ZKWatcherManager.newBuilder()
                 .name("test-register-unregister-watcher")
+                .zkc(null)
                 .statsLogger(NullStatsLogger.INSTANCE)
                 .build();
         String path = "/test-register-unregister-watcher";


Mime
View raw message