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.util; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertFalse; 022import static org.junit.Assert.assertTrue; 023 024import java.io.IOException; 025import java.util.ArrayList; 026import java.util.List; 027import org.apache.hadoop.hbase.HBaseClassTestRule; 028import org.apache.hadoop.hbase.HBaseTestingUtility; 029import org.apache.hadoop.hbase.HConstants; 030import org.apache.hadoop.hbase.MetaTableAccessor; 031import org.apache.hadoop.hbase.TableName; 032import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 033import org.apache.hadoop.hbase.client.Put; 034import org.apache.hadoop.hbase.client.RegionInfo; 035import org.apache.hadoop.hbase.client.RegionInfoBuilder; 036import org.apache.hadoop.hbase.client.Result; 037import org.apache.hadoop.hbase.client.ResultScanner; 038import org.apache.hadoop.hbase.client.Scan; 039import org.apache.hadoop.hbase.client.Table; 040import org.apache.hadoop.hbase.client.TableDescriptor; 041import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 042import org.apache.hadoop.hbase.master.RegionState; 043import org.apache.hadoop.hbase.replication.ReplicationException; 044import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; 045import org.apache.hadoop.hbase.replication.ReplicationQueueStorage; 046import org.apache.hadoop.hbase.replication.ReplicationStorageFactory; 047import org.apache.hadoop.hbase.testclassification.MediumTests; 048import org.apache.hadoop.hbase.testclassification.ReplicationTests; 049import org.apache.hadoop.hbase.util.hbck.HbckTestingUtil; 050import org.junit.AfterClass; 051import org.junit.BeforeClass; 052import org.junit.ClassRule; 053import org.junit.Test; 054import org.junit.experimental.categories.Category; 055 056import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap; 057 058@Category({ ReplicationTests.class, MediumTests.class }) 059public class TestHBaseFsckCleanReplicationBarriers { 060 @ClassRule 061 public static final HBaseClassTestRule CLASS_RULE = 062 HBaseClassTestRule.forClass(TestHBaseFsckCleanReplicationBarriers.class); 063 064 private static final HBaseTestingUtility UTIL = new HBaseTestingUtility(); 065 066 private static String PEER_1 = "1", PEER_2 = "2"; 067 068 private static ReplicationQueueStorage QUEUE_STORAGE; 069 070 private static String WAL_FILE_NAME = "test.wal"; 071 072 private static String TABLE_NAME = "test"; 073 074 private static String COLUMN_FAMILY = "info"; 075 076 @BeforeClass 077 public static void setUp() throws Exception { 078 UTIL.startMiniCluster(1); 079 QUEUE_STORAGE = ReplicationStorageFactory.getReplicationQueueStorage(UTIL.getZooKeeperWatcher(), 080 UTIL.getConfiguration()); 081 createPeer(); 082 QUEUE_STORAGE.addWAL(UTIL.getMiniHBaseCluster().getRegionServer(0).getServerName(), PEER_1, 083 WAL_FILE_NAME); 084 QUEUE_STORAGE.addWAL(UTIL.getMiniHBaseCluster().getRegionServer(0).getServerName(), PEER_2, 085 WAL_FILE_NAME); 086 } 087 088 @AfterClass 089 public static void tearDown() throws Exception { 090 UTIL.shutdownMiniCluster(); 091 } 092 093 @Test 094 public void testCleanReplicationBarrierWithNonExistTable() 095 throws ClassNotFoundException, IOException { 096 TableName tableName = TableName.valueOf(TABLE_NAME + "_non"); 097 boolean cleaned = HbckTestingUtil.cleanReplicationBarrier(UTIL.getConfiguration(), tableName); 098 assertFalse(cleaned); 099 } 100 101 @Test 102 public void testCleanReplicationBarrierWithDeletedTable() throws Exception { 103 TableName tableName = TableName.valueOf(TABLE_NAME + "_deleted"); 104 List<RegionInfo> regionInfos = new ArrayList<>(); 105 // only write some barriers into meta table 106 107 for (int i = 0; i < 110; i++) { 108 RegionInfo regionInfo = RegionInfoBuilder.newBuilder(tableName).setStartKey(Bytes.toBytes(i)) 109 .setEndKey(Bytes.toBytes(i + 1)).build(); 110 regionInfos.add(regionInfo); 111 addStateAndBarrier(regionInfo, RegionState.State.OPEN, 10, 100); 112 updatePushedSeqId(regionInfo, 10); 113 assertEquals("check if there is lastPushedId", 10, 114 QUEUE_STORAGE.getLastSequenceId(regionInfo.getEncodedName(), PEER_1)); 115 assertEquals("check if there is lastPushedId", 10, 116 QUEUE_STORAGE.getLastSequenceId(regionInfo.getEncodedName(), PEER_2)); 117 } 118 Scan barrierScan = new Scan(); 119 barrierScan.setCaching(100); 120 barrierScan.addFamily(HConstants.REPLICATION_BARRIER_FAMILY); 121 barrierScan 122 .withStartRow( 123 MetaTableAccessor.getTableStartRowForMeta(tableName, MetaTableAccessor.QueryType.REGION)) 124 .withStopRow( 125 MetaTableAccessor.getTableStopRowForMeta(tableName, MetaTableAccessor.QueryType.REGION)); 126 Result result; 127 try (ResultScanner scanner = 128 MetaTableAccessor.getMetaHTable(UTIL.getConnection()).getScanner(barrierScan)) { 129 while ((result = scanner.next()) != null) { 130 assertTrue(MetaTableAccessor.getReplicationBarriers(result).length > 0); 131 } 132 } 133 boolean cleaned = HbckTestingUtil.cleanReplicationBarrier(UTIL.getConfiguration(), tableName); 134 assertTrue(cleaned); 135 for (RegionInfo regionInfo : regionInfos) { 136 assertEquals("check if there is lastPushedId", -1, 137 QUEUE_STORAGE.getLastSequenceId(regionInfo.getEncodedName(), PEER_1)); 138 assertEquals("check if there is lastPushedId", -1, 139 QUEUE_STORAGE.getLastSequenceId(regionInfo.getEncodedName(), PEER_2)); 140 } 141 cleaned = HbckTestingUtil.cleanReplicationBarrier(UTIL.getConfiguration(), tableName); 142 assertFalse(cleaned); 143 for (RegionInfo region : regionInfos) { 144 assertEquals(0, MetaTableAccessor.getReplicationBarrier(UTIL.getConnection(), 145 region.getRegionName()).length); 146 } 147 } 148 149 @Test 150 public void testCleanReplicationBarrierWithExistTable() throws Exception { 151 TableName tableName = TableName.valueOf(TABLE_NAME); 152 String cf = COLUMN_FAMILY; 153 TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName) 154 .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build()) 155 .setReplicationScope(HConstants.REPLICATION_SCOPE_LOCAL).build(); 156 UTIL.createTable(tableDescriptor, Bytes.split(Bytes.toBytes(1), Bytes.toBytes(256), 123)); 157 assertTrue(UTIL.getAdmin().getRegions(tableName).size() > 0); 158 for (RegionInfo region : UTIL.getAdmin().getRegions(tableName)) { 159 addStateAndBarrier(region, RegionState.State.OFFLINE, 10, 100); 160 updatePushedSeqId(region, 10); 161 assertEquals("check if there is lastPushedId", 10, 162 QUEUE_STORAGE.getLastSequenceId(region.getEncodedName(), PEER_1)); 163 assertEquals("check if there is lastPushedId", 10, 164 QUEUE_STORAGE.getLastSequenceId(region.getEncodedName(), PEER_2)); 165 } 166 boolean cleaned = HbckTestingUtil.cleanReplicationBarrier(UTIL.getConfiguration(), tableName); 167 assertTrue(cleaned); 168 for (RegionInfo region : UTIL.getAdmin().getRegions(tableName)) { 169 assertEquals("check if there is lastPushedId", -1, 170 QUEUE_STORAGE.getLastSequenceId(region.getEncodedName(), PEER_1)); 171 assertEquals("check if there is lastPushedId", -1, 172 QUEUE_STORAGE.getLastSequenceId(region.getEncodedName(), PEER_2)); 173 } 174 cleaned = HbckTestingUtil.cleanReplicationBarrier(UTIL.getConfiguration(), tableName); 175 assertFalse(cleaned); 176 for (RegionInfo region : UTIL.getAdmin().getRegions(tableName)) { 177 assertEquals(0, MetaTableAccessor.getReplicationBarrier(UTIL.getConnection(), 178 region.getRegionName()).length); 179 } 180 } 181 182 public static void createPeer() throws IOException { 183 ReplicationPeerConfig rpc = ReplicationPeerConfig.newBuilder() 184 .setClusterKey(UTIL.getClusterKey() + "-test").setSerial(true).build(); 185 UTIL.getAdmin().addReplicationPeer(PEER_1, rpc); 186 UTIL.getAdmin().addReplicationPeer(PEER_2, rpc); 187 } 188 189 private void addStateAndBarrier(RegionInfo region, RegionState.State state, long... barriers) 190 throws IOException { 191 Put put = new Put(region.getRegionName(), EnvironmentEdgeManager.currentTime()); 192 if (state != null) { 193 put.addColumn(HConstants.CATALOG_FAMILY, HConstants.STATE_QUALIFIER, 194 Bytes.toBytes(state.name())); 195 } 196 for (int i = 0; i < barriers.length; i++) { 197 put.addColumn(HConstants.REPLICATION_BARRIER_FAMILY, HConstants.SEQNUM_QUALIFIER, 198 put.getTimeStamp() - barriers.length + i, Bytes.toBytes(barriers[i])); 199 } 200 try (Table table = UTIL.getConnection().getTable(TableName.META_TABLE_NAME)) { 201 table.put(put); 202 } 203 } 204 205 private void updatePushedSeqId(RegionInfo region, long seqId) throws ReplicationException { 206 QUEUE_STORAGE.setWALPosition(UTIL.getMiniHBaseCluster().getRegionServer(0).getServerName(), 207 PEER_1, WAL_FILE_NAME, 10, ImmutableMap.of(region.getEncodedName(), seqId)); 208 QUEUE_STORAGE.setWALPosition(UTIL.getMiniHBaseCluster().getRegionServer(0).getServerName(), 209 PEER_2, WAL_FILE_NAME, 10, ImmutableMap.of(region.getEncodedName(), seqId)); 210 } 211}