001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.replication.regionserver;
019
020import static org.junit.Assert.assertTrue;
021
022import org.apache.hadoop.hbase.HBaseClassTestRule;
023import org.apache.hadoop.hbase.Waiter.ExplainingPredicate;
024import org.apache.hadoop.hbase.client.Delete;
025import org.apache.hadoop.hbase.client.Get;
026import org.apache.hadoop.hbase.client.RegionInfoBuilder;
027import org.apache.hadoop.hbase.client.Table;
028import org.apache.hadoop.hbase.regionserver.HRegion;
029import org.apache.hadoop.hbase.regionserver.HRegionServer;
030import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
031import org.apache.hadoop.hbase.replication.SyncReplicationState;
032import org.apache.hadoop.hbase.replication.SyncReplicationTestBase;
033import org.apache.hadoop.hbase.testclassification.MediumTests;
034import org.apache.hadoop.hbase.testclassification.ReplicationTests;
035import org.apache.hadoop.hbase.util.Bytes;
036import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
037import org.junit.ClassRule;
038import org.junit.Test;
039import org.junit.experimental.categories.Category;
040
041@Category({ ReplicationTests.class, MediumTests.class })
042public class TestDrainReplicationQueuesForStandBy extends SyncReplicationTestBase {
043
044  @ClassRule
045  public static final HBaseClassTestRule CLASS_RULE =
046    HBaseClassTestRule.forClass(TestDrainReplicationQueuesForStandBy.class);
047
048  @Test
049  public void test() throws Exception {
050    UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
051      SyncReplicationState.STANDBY);
052    UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
053      SyncReplicationState.ACTIVE);
054    UTIL1.getAdmin().disableReplicationPeer(PEER_ID);
055    write(UTIL1, 0, 100);
056
057    HRegionServer rs = UTIL1.getRSForFirstRegionInTable(TABLE_NAME);
058    String walGroupId = AbstractFSWALProvider.getWALPrefixFromWALName(
059      ((AbstractFSWAL<?>) rs.getWAL(RegionInfoBuilder.newBuilder(TABLE_NAME).build()))
060        .getCurrentFileName().getName());
061    UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
062      SyncReplicationState.DOWNGRADE_ACTIVE);
063    // transit cluster2 to DA and cluster 1 to S
064    verify(UTIL2, 0, 100);
065
066    UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
067      SyncReplicationState.STANDBY);
068    // delete the original value, and then major compact
069    try (Table table = UTIL2.getConnection().getTable(TABLE_NAME)) {
070      for (int i = 0; i < 100; i++) {
071        table.delete(new Delete(Bytes.toBytes(i)));
072      }
073    }
074    UTIL2.flush(TABLE_NAME);
075    UTIL2.compact(TABLE_NAME, true);
076    // wait until the new values are replicated back to cluster1
077    HRegion region = rs.getRegions(TABLE_NAME).get(0);
078    UTIL1.waitFor(30000, new ExplainingPredicate<Exception>() {
079
080      @Override
081      public boolean evaluate() throws Exception {
082        return region.get(new Get(Bytes.toBytes(99))).isEmpty();
083      }
084
085      @Override
086      public String explainFailure() throws Exception {
087        return "Replication has not been catched up yet";
088      }
089    });
090    // transit cluster1 to DA and cluster2 to S, then we will start replicating from cluster1 to
091    // cluster2
092    UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
093      SyncReplicationState.DOWNGRADE_ACTIVE);
094    UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
095      SyncReplicationState.STANDBY);
096    UTIL1.getAdmin().enableReplicationPeer(PEER_ID);
097
098    // confirm that we will not replicate the old data which causes inconsistency
099    ReplicationSource source = (ReplicationSource) ((Replication) rs.getReplicationSourceService())
100      .getReplicationManager().getSource(PEER_ID);
101    UTIL1.waitFor(30000, new ExplainingPredicate<Exception>() {
102
103      @Override
104      public boolean evaluate() throws Exception {
105        return !source.workerThreads.containsKey(walGroupId);
106      }
107
108      @Override
109      public String explainFailure() throws Exception {
110        return "Replication has not been catched up yet";
111      }
112    });
113    HRegion region2 = UTIL2.getMiniHBaseCluster().getRegions(TABLE_NAME).get(0);
114    for (int i = 0; i < 100; i++) {
115      assertTrue(region2.get(new Get(Bytes.toBytes(i))).isEmpty());
116    }
117  }
118}