You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ap...@apache.org on 2018/05/23 23:47:14 UTC

[3/7] hbase git commit: HBASE-20597 Use a lock to serialize access to a shared reference to ZooKeeperWatcher in HBaseReplicationEndpoint

HBASE-20597 Use a lock to serialize access to a shared reference to ZooKeeperWatcher in HBaseReplicationEndpoint


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

Branch: refs/heads/branch-1.3
Commit: b50e149804afeeb91bc561d2e2ae1ca6fe33c7d3
Parents: 94001b3
Author: Andrew Purtell <ap...@apache.org>
Authored: Thu May 17 10:30:28 2018 -0700
Committer: Andrew Purtell <ap...@apache.org>
Committed: Wed May 23 16:46:01 2018 -0700

----------------------------------------------------------------------
 .../replication/HBaseReplicationEndpoint.java   | 43 ++++++++++++--------
 1 file changed, 27 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/b50e1498/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
index 6485e4a..39a3f31 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
@@ -43,21 +43,22 @@ import org.apache.zookeeper.KeeperException.SessionExpiredException;
  * target cluster is an HBase cluster.
  */
 @InterfaceAudience.Private
-@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="MT_CORRECTNESS",
-  justification="Thinks zkw needs to be synchronized access but should be fine as is.")
 public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
   implements Abortable {
 
   private static final Log LOG = LogFactory.getLog(HBaseReplicationEndpoint.class);
 
-  private ZooKeeperWatcher zkw = null; // FindBugs: MT_CORRECTNESS
+  private Object zkwLock = new Object();
+  private ZooKeeperWatcher zkw = null;
 
   private List<ServerName> regionServers = new ArrayList<ServerName>(0);
   private long lastRegionServerUpdate;
 
   protected void disconnect() {
-    if (zkw != null) {
-      zkw.close();
+    synchronized (zkwLock) {
+      if (zkw != null) {
+        zkw.close();
+      }
     }
   }
 
@@ -102,7 +103,9 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
   public synchronized UUID getPeerUUID() {
     UUID peerUUID = null;
     try {
-      peerUUID = ZKClusterId.getUUIDForCluster(zkw);
+      synchronized (zkwLock) {
+        peerUUID = ZKClusterId.getUUIDForCluster(zkw);
+      }
     } catch (KeeperException ke) {
       reconnect(ke);
     }
@@ -114,7 +117,9 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
    * @return zk connection
    */
   protected ZooKeeperWatcher getZkw() {
-    return zkw;
+    synchronized (zkwLock) {
+      return zkw;
+    }
   }
 
   /**
@@ -122,10 +127,14 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
    * @throws IOException If anything goes wrong connecting
    */
   void reloadZkWatcher() throws IOException {
-    if (zkw != null) zkw.close();
-    zkw = new ZooKeeperWatcher(ctx.getConfiguration(),
+    synchronized (zkwLock) {
+      if (zkw != null) {
+        zkw.close();
+      }
+      zkw = new ZooKeeperWatcher(ctx.getConfiguration(),
         "connection to cluster: " + ctx.getPeerId(), this);
-    getZkw().registerListener(new PeerRegionServerListener(this));
+      zkw.registerListener(new PeerRegionServerListener(this));
+    }
   }
 
   @Override
@@ -163,13 +172,15 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
    * for this peer cluster
    * @return list of addresses
    */
-  // Synchronize peer cluster connection attempts to avoid races and rate
-  // limit connections when multiple replication sources try to connect to
-  // the peer cluster. If the peer cluster is down we can get out of control
-  // over time.
-  public synchronized List<ServerName> getRegionServers() {
+  public List<ServerName> getRegionServers() {
     try {
-      setRegionServers(fetchSlavesAddresses(this.getZkw()));
+      // Synchronize peer cluster connection attempts to avoid races and rate
+      // limit connections when multiple replication sources try to connect to
+      // the peer cluster. If the peer cluster is down we can get out of control
+      // over time.
+      synchronized (zkwLock) {
+        setRegionServers(fetchSlavesAddresses(zkw));
+      }
     } catch (KeeperException ke) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Fetch slaves addresses failed", ke);