hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationTrackerZKImpl.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.replication;
  19
  20 import static org.junit.Assert.assertEquals;
  21 import static org.junit.Assert.fail;
  22
  23 import java.io.IOException;
  24 import java.util.List;
  25 import java.util.concurrent.atomic.AtomicInteger;
  26 import org.apache.hadoop.conf.Configuration;
  27 import org.apache.hadoop.fs.FileSystem;
  28 import org.apache.hadoop.hbase.ChoreService;
  29 import org.apache.hadoop.hbase.ClusterId;
  30 import org.apache.hadoop.hbase.CoordinatedStateManager;
  31 import org.apache.hadoop.hbase.HBaseClassTestRule;
  32 import org.apache.hadoop.hbase.HBaseTestingUtility;
  33 import org.apache.hadoop.hbase.HConstants;
  34 import org.apache.hadoop.hbase.Server;
  35 import org.apache.hadoop.hbase.ServerName;
  36 import org.apache.hadoop.hbase.client.AsyncClusterConnection;
  37 import org.apache.hadoop.hbase.client.Connection;
  38 import org.apache.hadoop.hbase.testclassification.MediumTests;
  39 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
  40 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
  41 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
  42 import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
  43 import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
  44 import org.apache.zookeeper.KeeperException;
  45 import org.junit.AfterClass;
  46 import org.junit.Before;
  47 import org.junit.BeforeClass;
  48 import org.junit.ClassRule;
  49 import org.junit.Test;
  50 import org.junit.experimental.categories.Category;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 /**
  55  * This class tests the ReplicationTrackerZKImpl class and ReplicationListener interface. One
  56  * MiniZKCluster is used throughout the entire class. The cluster is initialized with the creation
  57  * of the rsZNode. All other znode creation/initialization is handled by the replication state
  58  * interfaces (i.e. ReplicationPeers, etc.). Each test case in this class should ensure that the
  59  * MiniZKCluster is cleaned and returned to it's initial state (i.e. nothing but the rsZNode).
  60  */
  61 @Category({ReplicationTests.class, MediumTests.class})
  62 public class TestReplicationTrackerZKImpl {
  63
  64   @ClassRule
  65   public static final HBaseClassTestRule CLASS_RULE =
  66       HBaseClassTestRule.forClass(TestReplicationTrackerZKImpl.class);
  67
  68   private static final Logger LOG = LoggerFactory.getLogger(TestReplicationTrackerZKImpl.class);
  69
  70   private static Configuration conf;
  71   private static HBaseTestingUtility utility;
  72
  73   // Each one of the below variables are reinitialized before every test case
  74   private ZKWatcher zkw;
  75   private ReplicationPeers rp;
  76   private ReplicationTracker rt;
  77   private AtomicInteger rsRemovedCount;
  78   private String rsRemovedData;
  79
  80   @BeforeClass
  81   public static void setUpBeforeClass() throws Exception {
  82     utility = new HBaseTestingUtility();
  83     utility.startMiniZKCluster();
  84     conf = utility.getConfiguration();
  85     ZKWatcher zk = HBaseTestingUtility.getZooKeeperWatcher(utility);
  86     ZKUtil.createWithParents(zk, zk.getZNodePaths().rsZNode);
  87   }
  88
  89   @Before
  90   public void setUp() throws Exception {
  91     zkw = HBaseTestingUtility.getZooKeeperWatcher(utility);
  92     String fakeRs1 = ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode,
  93             "hostname1.example.org:1234");
  94     try {
  95       ZKClusterId.setClusterId(zkw, new ClusterId());
  96       rp = ReplicationFactory.getReplicationPeers(zkw, conf);
  97       rp.init();
  98       rt = ReplicationFactory.getReplicationTracker(zkw, new DummyServer(fakeRs1),
  99         new DummyServer(fakeRs1));
 100     } catch (Exception e) {
 101       fail("Exception during test setup: " + e);
 102     }
 103     rsRemovedCount = new AtomicInteger(0);
 104     rsRemovedData = "";
 105   }
 106
 107   @AfterClass
 108   public static void tearDownAfterClass() throws Exception {
 109     utility.shutdownMiniZKCluster();
 110   }
 111
 112   @Test
 113   public void testGetListOfRegionServers() throws Exception {
 114     // 0 region servers
 115     assertEquals(0, rt.getListOfRegionServers().size());
 116
 117     // 1 region server
 118     ZKUtil.createWithParents(zkw,
 119       ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname1.example.org:1234"));
 120     List<String> rss = rt.getListOfRegionServers();
 121     assertEquals(rss.toString(), 1, rss.size());
 122
 123     // 2 region servers
 124     ZKUtil.createWithParents(zkw,
 125       ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname2.example.org:1234"));
 126     rss = rt.getListOfRegionServers();
 127     assertEquals(rss.toString(), 2, rss.size());
 128
 129     // 1 region server
 130     ZKUtil.deleteNode(zkw,
 131       ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname2.example.org:1234"));
 132     rss = rt.getListOfRegionServers();
 133     assertEquals(1, rss.size());
 134
 135     // 0 region server
 136     ZKUtil.deleteNode(zkw,
 137       ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname1.example.org:1234"));
 138     rss = rt.getListOfRegionServers();
 139     assertEquals(rss.toString(), 0, rss.size());
 140   }
 141
 142   @Test
 143   public void testRegionServerRemovedEvent() throws Exception {
 144     ZKUtil.createAndWatch(zkw,
 145       ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname2.example.org:1234"),
 146       HConstants.EMPTY_BYTE_ARRAY);
 147     rt.registerListener(new DummyReplicationListener());
 148     // delete one
 149     ZKUtil.deleteNode(zkw,
 150       ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname2.example.org:1234"));
 151     // wait for event
 152     while (rsRemovedCount.get() < 1) {
 153       Thread.sleep(5);
 154     }
 155     assertEquals("hostname2.example.org:1234", rsRemovedData);
 156   }
 157
 158   @Test
 159   public void testPeerNameControl() throws Exception {
 160     int exists = 0;
 161     rp.getPeerStorage().addPeer("6",
 162       ReplicationPeerConfig.newBuilder().setClusterKey(utility.getClusterKey()).build(), true,
 163       SyncReplicationState.NONE);
 164
 165     try {
 166       rp.getPeerStorage().addPeer("6",
 167         ReplicationPeerConfig.newBuilder().setClusterKey(utility.getClusterKey()).build(), true,
 168         SyncReplicationState.NONE);
 169     } catch (ReplicationException e) {
 170       if (e.getCause() instanceof KeeperException.NodeExistsException) {
 171         exists++;
 172       }
 173     }
 174
 175     assertEquals(1, exists);
 176
 177     // clean up
 178     rp.getPeerStorage().removePeer("6");
 179   }
 180
 181   private class DummyReplicationListener implements ReplicationListener {
 182
 183     @Override
 184     public void regionServerRemoved(String regionServer) {
 185       rsRemovedData = regionServer;
 186       rsRemovedCount.getAndIncrement();
 187       LOG.debug("Received regionServerRemoved event: " + regionServer);
 188     }
 189   }
 190
 191   private class DummyServer implements Server {
 192     private String serverName;
 193     private boolean isAborted = false;
 194     private boolean isStopped = false;
 195
 196     public DummyServer(String serverName) {
 197       this.serverName = serverName;
 198     }
 199
 200     @Override
 201     public Configuration getConfiguration() {
 202       return conf;
 203     }
 204
 205     @Override
 206     public ZKWatcher getZooKeeper() {
 207       return zkw;
 208     }
 209
 210     @Override
 211     public CoordinatedStateManager getCoordinatedStateManager() {
 212       return null;
 213     }
 214
 215     @Override
 216     public Connection getConnection() {
 217       return null;
 218     }
 219
 220     @Override
 221     public ServerName getServerName() {
 222       return ServerName.valueOf(this.serverName);
 223     }
 224
 225     @Override
 226     public void abort(String why, Throwable e) {
 227       LOG.info("Aborting " + serverName);
 228       this.isAborted = true;
 229     }
 230
 231     @Override
 232     public boolean isAborted() {
 233       return this.isAborted;
 234     }
 235
 236     @Override
 237     public void stop(String why) {
 238       this.isStopped = true;
 239     }
 240
 241     @Override
 242     public boolean isStopped() {
 243       return this.isStopped;
 244     }
 245
 246     @Override
 247     public ChoreService getChoreService() {
 248       return null;
 249     }
 250
 251     @Override
 252     public FileSystem getFileSystem() {
 253       return null;
 254     }
 255
 256     @Override
 257     public boolean isStopping() {
 258       return false;
 259     }
 260
 261     @Override
 262     public Connection createConnection(Configuration conf) throws IOException {
 263       return null;
 264     }
 265
 266     @Override
 267     public AsyncClusterConnection getAsyncClusterConnection() {
 268       return null;
 269     }
 270   }
 271 }