hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSyncReplicationNewRSJoinBetweenRefreshes.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.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.PeerSyncReplicationStateTransitionState.REOPEN_ALL_REGIONS_IN_PEER_VALUE;
  21 import static org.junit.Assert.assertEquals;
  22
  23 import java.io.IOException;
  24 import java.io.UncheckedIOException;
  25 import java.util.Optional;
  26 import java.util.concurrent.CountDownLatch;
  27 import org.apache.hadoop.hbase.HBaseClassTestRule;
  28 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
  29 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
  30 import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessor;
  31 import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment;
  32 import org.apache.hadoop.hbase.coprocessor.RegionServerObserver;
  33 import org.apache.hadoop.hbase.master.replication.TransitPeerSyncReplicationStateProcedure;
  34 import org.apache.hadoop.hbase.testclassification.LargeTests;
  35 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
  36 import org.junit.BeforeClass;
  37 import org.junit.ClassRule;
  38 import org.junit.Test;
  39 import org.junit.experimental.categories.Category;
  40
  41 /**
  42  * Testcase for HBASE-21441.
  43  */
  44 @Category({ ReplicationTests.class, LargeTests.class })
  45 public class TestSyncReplicationNewRSJoinBetweenRefreshes extends SyncReplicationTestBase {
  46
  47   @ClassRule
  48   public static final HBaseClassTestRule CLASS_RULE =
  49     HBaseClassTestRule.forClass(TestSyncReplicationNewRSJoinBetweenRefreshes.class);
  50
  51   private static boolean HALT;
  52
  53   private static CountDownLatch ARRIVE;
  54
  55   private static CountDownLatch RESUME;
  56
  57   public static final class HaltCP implements RegionServerObserver, RegionServerCoprocessor {
  58
  59     @Override
  60     public Optional<RegionServerObserver> getRegionServerObserver() {
  61       return Optional.of(this);
  62     }
  63
  64     @Override
  65     public void postExecuteProcedures(ObserverContext<RegionServerCoprocessorEnvironment> ctx)
  66         throws IOException {
  67       synchronized (HaltCP.class) {
  68         if (!HALT) {
  69           return;
  70         }
  71         UTIL1.getMiniHBaseCluster().getMaster().getProcedures().stream()
  72           .filter(p -> p instanceof TransitPeerSyncReplicationStateProcedure)
  73           .filter(p -> !p.isFinished()).map(p -> (TransitPeerSyncReplicationStateProcedure) p)
  74           .findFirst().ifPresent(proc -> {
  75             // this is the next state of REFRESH_PEER_SYNC_REPLICATION_STATE_ON_RS_BEGIN_VALUE
  76             if (proc.getCurrentStateId() == REOPEN_ALL_REGIONS_IN_PEER_VALUE) {
  77               // tell the main thread to start a new region server
  78               ARRIVE.countDown();
  79               try {
  80                 // wait for the region server to online
  81                 RESUME.await();
  82               } catch (InterruptedException e) {
  83                 throw new RuntimeException(e);
  84               }
  85               HALT = false;
  86             }
  87           });
  88       }
  89     }
  90   }
  91
  92   @BeforeClass
  93   public static void setUp() throws Exception {
  94     UTIL1.getConfiguration().setClass(CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY,
  95       HaltCP.class, RegionServerObserver.class);
  96     SyncReplicationTestBase.setUp();
  97   }
  98
  99   @Test
 100   public void test() throws IOException, InterruptedException {
 101     UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
 102       SyncReplicationState.STANDBY);
 103     UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
 104       SyncReplicationState.ACTIVE);
 105
 106     ARRIVE = new CountDownLatch(1);
 107     RESUME = new CountDownLatch(1);
 108     HALT = true;
 109     Thread t = new Thread(() -> {
 110       try {
 111         UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
 112           SyncReplicationState.DOWNGRADE_ACTIVE);
 113       } catch (IOException e) {
 114         throw new UncheckedIOException(e);
 115       }
 116     });
 117     t.start();
 118     ARRIVE.await();
 119     UTIL1.getMiniHBaseCluster().startRegionServer();
 120     RESUME.countDown();
 121     t.join();
 122     assertEquals(SyncReplicationState.DOWNGRADE_ACTIVE,
 123       UTIL1.getAdmin().getReplicationPeerSyncReplicationState(PEER_ID));
 124   }
 125 }