hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSyncReplicationStandbyKillRS.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.assertFalse;
  21 import static org.junit.Assert.assertTrue;
  22
  23 import java.util.List;
  24
  25 import org.apache.hadoop.fs.Path;
  26 import org.apache.hadoop.hbase.HBaseClassTestRule;
  27 import org.apache.hadoop.hbase.ServerName;
  28 import org.apache.hadoop.hbase.master.MasterFileSystem;
  29 import org.apache.hadoop.hbase.master.ServerManager;
  30 import org.apache.hadoop.hbase.testclassification.LargeTests;
  31 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
  32 import org.apache.hadoop.hbase.util.JVMClusterUtil;
  33 import org.junit.ClassRule;
  34 import org.junit.Test;
  35 import org.junit.experimental.categories.Category;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 @Category({ ReplicationTests.class, LargeTests.class })
  40 public class TestSyncReplicationStandbyKillRS extends SyncReplicationTestBase {
  41
  42   private static final Logger LOG =
  43       LoggerFactory.getLogger(TestSyncReplicationStandbyKillRS.class);
  44
  45   private final long SLEEP_TIME = 1000;
  46
  47   private final int COUNT = 1000;
  48
  49   @ClassRule
  50   public static final HBaseClassTestRule CLASS_RULE =
  51       HBaseClassTestRule.forClass(TestSyncReplicationStandbyKillRS.class);
  52
  53   @Test
  54   public void testStandbyKillRegionServer() throws Exception {
  55     MasterFileSystem mfs = UTIL2.getHBaseCluster().getMaster().getMasterFileSystem();
  56     Path remoteWALDir = getRemoteWALDir(mfs, PEER_ID);
  57     assertFalse(mfs.getWALFileSystem().exists(remoteWALDir));
  58     UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
  59         SyncReplicationState.STANDBY);
  60     assertTrue(mfs.getWALFileSystem().exists(remoteWALDir));
  61     UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
  62         SyncReplicationState.ACTIVE);
  63
  64     // Disable async replication and write data, then shutdown
  65     UTIL1.getAdmin().disableReplicationPeer(PEER_ID);
  66     write(UTIL1, 0, COUNT);
  67     UTIL1.shutdownMiniCluster();
  68
  69     JVMClusterUtil.MasterThread activeMaster = UTIL2.getMiniHBaseCluster().getMasterThread();
  70     Thread t = new Thread(() -> {
  71       try {
  72         List<JVMClusterUtil.RegionServerThread> regionServers =
  73             UTIL2.getMiniHBaseCluster().getLiveRegionServerThreads();
  74         for (JVMClusterUtil.RegionServerThread rst : regionServers) {
  75           ServerName serverName = rst.getRegionServer().getServerName();
  76           rst.getRegionServer().stop("Stop RS for test");
  77           waitForRSShutdownToStartAndFinish(activeMaster, serverName);
  78           JVMClusterUtil.RegionServerThread restarted =
  79               UTIL2.getMiniHBaseCluster().startRegionServer();
  80           restarted.waitForServerOnline();
  81         }
  82       } catch (Exception e) {
  83         LOG.error("Failed to kill RS", e);
  84       }
  85     });
  86     t.start();
  87
  88     // Transit standby to DA to replay logs
  89     try {
  90       UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
  91         SyncReplicationState.DOWNGRADE_ACTIVE);
  92     } catch (Exception e) {
  93       LOG.error("Failed to transit standby cluster to " + SyncReplicationState.DOWNGRADE_ACTIVE, e);
  94     }
  95
  96     while (UTIL2.getAdmin().getReplicationPeerSyncReplicationState(PEER_ID)
  97         != SyncReplicationState.DOWNGRADE_ACTIVE) {
  98       Thread.sleep(SLEEP_TIME);
  99     }
 100     verify(UTIL2, 0, COUNT);
 101   }
 102
 103   private void waitForRSShutdownToStartAndFinish(JVMClusterUtil.MasterThread activeMaster,
 104       ServerName serverName) throws InterruptedException {
 105     ServerManager sm = activeMaster.getMaster().getServerManager();
 106     // First wait for it to be in dead list
 107     while (!sm.getDeadServers().isDeadServer(serverName)) {
 108       LOG.debug("Waiting for [" + serverName + "] to be listed as dead in master");
 109       Thread.sleep(SLEEP_TIME);
 110     }
 111     LOG.debug("Server [" + serverName + "] marked as dead, waiting for it to " +
 112         "finish dead processing");
 113     while (sm.areDeadServersInProgress()) {
 114       LOG.debug("Server [" + serverName + "] still being processed, waiting");
 115       Thread.sleep(SLEEP_TIME);
 116     }
 117     LOG.debug("Server [" + serverName + "] done with server shutdown processing");
 118   }
 119 }