HBASE-19923 Reset peer state and config when refresh replication source failed
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/3db22e6d Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/3db22e6d Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/3db22e6d Branch: refs/heads/HBASE-19397-branch-2 Commit: 3db22e6dda5d2bb652dde91064ac3d134185b40f Parents: c1b2ad6 Author: Guanghao Zhang <zg...@apache.org> Authored: Tue Feb 6 14:58:39 2018 +0800 Committer: zhangduo <zhang...@apache.org> Committed: Fri Mar 2 15:00:51 2018 +0800 ---------------------------------------------------------------------- .../hbase/replication/ReplicationPeerImpl.java | 4 ++-- .../regionserver/PeerProcedureHandlerImpl.java | 24 ++++++++++++++++---- 2 files changed, 22 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/3db22e6d/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerImpl.java ---------------------------------------------------------------------- diff --git a/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerImpl.java b/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerImpl.java index 604e0bb..d656466 100644 --- a/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerImpl.java +++ b/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerImpl.java @@ -54,11 +54,11 @@ public class ReplicationPeerImpl implements ReplicationPeer { this.peerConfigListeners = new ArrayList<>(); } - void setPeerState(boolean enabled) { + public void setPeerState(boolean enabled) { this.peerState = enabled ? PeerState.ENABLED : PeerState.DISABLED; } - void setPeerConfig(ReplicationPeerConfig peerConfig) { + public void setPeerConfig(ReplicationPeerConfig peerConfig) { this.peerConfig = peerConfig; peerConfigListeners.forEach(listener -> listener.peerConfigUpdated(peerConfig)); } http://git-wip-us.apache.org/repos/asf/hbase/blob/3db22e6d/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/PeerProcedureHandlerImpl.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/PeerProcedureHandlerImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/PeerProcedureHandlerImpl.java index ce8fdae..a02d181 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/PeerProcedureHandlerImpl.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/PeerProcedureHandlerImpl.java @@ -62,18 +62,26 @@ public class PeerProcedureHandlerImpl implements PeerProcedureHandler { private void refreshPeerState(String peerId) throws ReplicationException, IOException { PeerState newState; Lock peerLock = peersLock.acquireLock(peerId); + ReplicationPeerImpl peer = null; + PeerState oldState = null; + boolean success = false; try { - ReplicationPeerImpl peer = replicationSourceManager.getReplicationPeers().getPeer(peerId); + peer = replicationSourceManager.getReplicationPeers().getPeer(peerId); if (peer == null) { throw new ReplicationException("Peer with id=" + peerId + " is not cached."); } - PeerState oldState = peer.getPeerState(); + oldState = peer.getPeerState(); newState = replicationSourceManager.getReplicationPeers().refreshPeerState(peerId); // RS need to start work with the new replication state change if (oldState.equals(PeerState.ENABLED) && newState.equals(PeerState.DISABLED)) { replicationSourceManager.refreshSources(peerId); } + success = true; } finally { + if (!success && peer != null) { + // Reset peer state if refresh source failed + peer.setPeerState(oldState.equals(PeerState.ENABLED)); + } peerLock.unlock(); } } @@ -91,19 +99,27 @@ public class PeerProcedureHandlerImpl implements PeerProcedureHandler { @Override public void updatePeerConfig(String peerId) throws ReplicationException, IOException { Lock peerLock = peersLock.acquireLock(peerId); + ReplicationPeerImpl peer = null; + ReplicationPeerConfig oldConfig = null; + boolean success = false; try { - ReplicationPeerImpl peer = replicationSourceManager.getReplicationPeers().getPeer(peerId); + peer = replicationSourceManager.getReplicationPeers().getPeer(peerId); if (peer == null) { throw new ReplicationException("Peer with id=" + peerId + " is not cached."); } - ReplicationPeerConfig oldConfig = peer.getPeerConfig(); + oldConfig = peer.getPeerConfig(); ReplicationPeerConfig newConfig = replicationSourceManager.getReplicationPeers().refreshPeerConfig(peerId); // RS need to start work with the new replication config change if (!ReplicationUtils.isKeyConfigEqual(oldConfig, newConfig)) { replicationSourceManager.refreshSources(peerId); } + success = true; } finally { + if (!success && peer != null) { + // Reset peer config if refresh source failed + peer.setPeerConfig(oldConfig); + } peerLock.unlock(); } }
