Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

This commit is contained in:
Andrew Purtell 2018-05-17 10:30:28 -07:00
parent 3a805074a2
commit 9fbce1668b
1 changed files with 27 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
View File

@ -43,23 +43,24 @@ import org.slf4j.LoggerFactory;
* target cluster is an HBase cluster. * target cluster is an HBase cluster.
*/ */
@InterfaceAudience.Private @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 public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
implements Abortable { implements Abortable {
private static final Logger LOG = LoggerFactory.getLogger(HBaseReplicationEndpoint.class); private static final Logger LOG = LoggerFactory.getLogger(HBaseReplicationEndpoint.class);
private ZKWatcher zkw = null; // FindBugs: MT_CORRECTNESS private Object zkwLock = new Object();
private ZKWatcher zkw = null;
private List<ServerName> regionServers = new ArrayList<>(0); private List<ServerName> regionServers = new ArrayList<>(0);
private long lastRegionServerUpdate; private long lastRegionServerUpdate;
protected void disconnect() { protected void disconnect() {
synchronized (zkwLock) {
if (zkw != null) { if (zkw != null) {
zkw.close(); zkw.close();
} }
} }
}
/** /**
* A private method used to re-establish a zookeeper session with a peer cluster. * A private method used to re-establish a zookeeper session with a peer cluster.
@ -112,7 +113,9 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
public synchronized UUID getPeerUUID() { public synchronized UUID getPeerUUID() {
UUID peerUUID = null; UUID peerUUID = null;
try { try {
synchronized (zkwLock) {
peerUUID = ZKClusterId.getUUIDForCluster(zkw); peerUUID = ZKClusterId.getUUIDForCluster(zkw);
}
} catch (KeeperException ke) { } catch (KeeperException ke) {
reconnect(ke); reconnect(ke);
} }
@ -124,18 +127,24 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
* @return zk connection * @return zk connection
*/ */
protected ZKWatcher getZkw() { protected ZKWatcher getZkw() {
synchronized (zkwLock) {
return zkw; return zkw;
} }
}
/** /**
* Closes the current ZKW (if not null) and creates a new one * Closes the current ZKW (if not null) and creates a new one
* @throws IOException If anything goes wrong connecting * @throws IOException If anything goes wrong connecting
*/ */
void reloadZkWatcher() throws IOException { void reloadZkWatcher() throws IOException {
if (zkw != null) zkw.close(); synchronized (zkwLock) {
if (zkw != null) {
zkw.close();
}
zkw = new ZKWatcher(ctx.getConfiguration(), zkw = new ZKWatcher(ctx.getConfiguration(),
"connection to cluster: " + ctx.getPeerId(), this); "connection to cluster: " + ctx.getPeerId(), this);
getZkw().registerListener(new PeerRegionServerListener(this)); zkw.registerListener(new PeerRegionServerListener(this));
}
} }
@Override @Override
@ -173,13 +182,15 @@ public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
* for this peer cluster * for this peer cluster
* @return list of addresses * @return list of addresses
*/ */
public List<ServerName> getRegionServers() {
try {
// Synchronize peer cluster connection attempts to avoid races and rate // Synchronize peer cluster connection attempts to avoid races and rate
// limit connections when multiple replication sources try to connect to // 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 // the peer cluster. If the peer cluster is down we can get out of control
// over time. // over time.
public synchronized List<ServerName> getRegionServers() { synchronized (zkwLock) {
try { setRegionServers(fetchSlavesAddresses(zkw));
setRegionServers(fetchSlavesAddresses(this.getZkw())); }
} catch (KeeperException ke) { } catch (KeeperException ke) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Fetch slaves addresses failed", ke); LOG.debug("Fetch slaves addresses failed", ke);