HBASE-19943 Only allow removing sync replication peer which is in DA state
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/31f1752c Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/31f1752c Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/31f1752c Branch: refs/heads/HBASE-19064 Commit: 31f1752c3e2294f698c83b0a98a64f7a12548ca7 Parents: 41cf435 Author: huzheng <open...@gmail.com> Authored: Thu Mar 1 18:34:02 2018 +0800 Committer: zhangduo <zhang...@apache.org> Committed: Wed Mar 7 18:14:00 2018 +0800 ---------------------------------------------------------------------- .../replication/ReplicationPeerManager.java | 11 ++++ .../hbase/wal/SyncReplicationWALProvider.java | 2 +- .../replication/TestReplicationAdmin.java | 63 ++++++++++++++++++++ 3 files changed, 75 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/31f1752c/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java index d80e9a4..6698aa5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java @@ -120,8 +120,19 @@ public class ReplicationPeerManager { return desc; } + private void checkPeerInDAStateIfSyncReplication(String peerId) throws DoNotRetryIOException { + ReplicationPeerDescription desc = peers.get(peerId); + if (desc != null && desc.getPeerConfig().isSyncReplication() + && !SyncReplicationState.DOWNGRADE_ACTIVE.equals(desc.getSyncReplicationState())) { + throw new DoNotRetryIOException("Couldn't remove synchronous replication peer with state=" + + desc.getSyncReplicationState() + + ", Transit the synchronous replication state to be DOWNGRADE_ACTIVE firstly."); + } + } + public void preRemovePeer(String peerId) throws DoNotRetryIOException { checkPeerExists(peerId); + checkPeerInDAStateIfSyncReplication(peerId); } public void preEnablePeer(String peerId) throws DoNotRetryIOException { http://git-wip-us.apache.org/repos/asf/hbase/blob/31f1752c/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/SyncReplicationWALProvider.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/SyncReplicationWALProvider.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/SyncReplicationWALProvider.java index ac4b4cd..282aa21 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/SyncReplicationWALProvider.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/SyncReplicationWALProvider.java @@ -142,7 +142,7 @@ public class SyncReplicationWALProvider implements WALProvider, PeerActionListen @Override public WAL getWAL(RegionInfo region) throws IOException { if (region == null) { - return provider.getWAL(region); + return provider.getWAL(null); } Optional<Pair<String, String>> peerIdAndRemoteWALDir = peerInfoProvider.getPeerIdAndRemoteWALDir(region); http://git-wip-us.apache.org/repos/asf/hbase/blob/31f1752c/hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java index 0ad476f..486ab51 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java @@ -254,6 +254,62 @@ public class TestReplicationAdmin { } @Test + public void testRemovePeerWithNonDAState() throws Exception { + TableName tableName = TableName.valueOf(name.getMethodName()); + TEST_UTIL.createTable(tableName, Bytes.toBytes("family")); + ReplicationPeerConfigBuilder builder = ReplicationPeerConfig.newBuilder(); + + String rootDir = "hdfs://srv1:9999/hbase"; + builder.setClusterKey(KEY_ONE); + builder.setRemoteWALDir(rootDir); + builder.setReplicateAllUserTables(false); + Map<TableName, List<String>> tableCfs = new HashMap<>(); + tableCfs.put(tableName, new ArrayList<>()); + builder.setTableCFsMap(tableCfs); + hbaseAdmin.addReplicationPeer(ID_ONE, builder.build()); + assertEquals(SyncReplicationState.DOWNGRADE_ACTIVE, + hbaseAdmin.getReplicationPeerSyncReplicationState(ID_ONE)); + + // Transit sync replication state to ACTIVE. + hbaseAdmin.transitReplicationPeerSyncReplicationState(ID_ONE, SyncReplicationState.ACTIVE); + assertEquals(SyncReplicationState.ACTIVE, + hbaseAdmin.getReplicationPeerSyncReplicationState(ID_ONE)); + + try { + hbaseAdmin.removeReplicationPeer(ID_ONE); + fail("Can't remove a synchronous replication peer with state=ACTIVE"); + } catch (IOException e) { + // OK + } + + // Transit sync replication state to DA + hbaseAdmin.transitReplicationPeerSyncReplicationState(ID_ONE, + SyncReplicationState.DOWNGRADE_ACTIVE); + assertEquals(SyncReplicationState.DOWNGRADE_ACTIVE, + hbaseAdmin.getReplicationPeerSyncReplicationState(ID_ONE)); + // Transit sync replication state to STANDBY + hbaseAdmin.transitReplicationPeerSyncReplicationState(ID_ONE, SyncReplicationState.STANDBY); + assertEquals(SyncReplicationState.STANDBY, + hbaseAdmin.getReplicationPeerSyncReplicationState(ID_ONE)); + + try { + hbaseAdmin.removeReplicationPeer(ID_ONE); + fail("Can't remove a synchronous replication peer with state=STANDBY"); + } catch (IOException e) { + // OK + } + + // Transit sync replication state to DA + hbaseAdmin.transitReplicationPeerSyncReplicationState(ID_ONE, + SyncReplicationState.DOWNGRADE_ACTIVE); + assertEquals(SyncReplicationState.DOWNGRADE_ACTIVE, + hbaseAdmin.getReplicationPeerSyncReplicationState(ID_ONE)); + + hbaseAdmin.removeReplicationPeer(ID_ONE); + assertEquals(0, hbaseAdmin.listReplicationPeers().size()); + } + + @Test public void testAddPeerWithState() throws Exception { ReplicationPeerConfig rpc1 = new ReplicationPeerConfig(); rpc1.setClusterKey(KEY_ONE); @@ -1072,5 +1128,12 @@ public class TestReplicationAdmin { } catch (Exception e) { // OK } + hbaseAdmin.transitReplicationPeerSyncReplicationState(ID_SECOND, + SyncReplicationState.DOWNGRADE_ACTIVE); + assertEquals(SyncReplicationState.DOWNGRADE_ACTIVE, + hbaseAdmin.getReplicationPeerSyncReplicationState(ID_SECOND)); + hbaseAdmin.removeReplicationPeer(ID_ONE); + hbaseAdmin.removeReplicationPeer(ID_SECOND); + assertEquals(0, hbaseAdmin.listReplicationPeers().size()); } }
