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.regionserver; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertFalse; 022import static org.junit.Assert.assertTrue; 023import static org.junit.Assert.fail; 024 025import java.io.IOException; 026import java.util.ArrayList; 027import java.util.Arrays; 028import java.util.List; 029import java.util.Objects; 030import java.util.concurrent.ThreadLocalRandom; 031import java.util.concurrent.atomic.AtomicBoolean; 032import java.util.stream.Collectors; 033import org.apache.hadoop.conf.Configuration; 034import org.apache.hadoop.fs.FileSystem; 035import org.apache.hadoop.fs.Path; 036import org.apache.hadoop.hbase.CatalogFamilyFormat; 037import org.apache.hadoop.hbase.HBaseClassTestRule; 038import org.apache.hadoop.hbase.HBaseTestingUtil; 039import org.apache.hadoop.hbase.MetaTableAccessor; 040import org.apache.hadoop.hbase.ServerName; 041import org.apache.hadoop.hbase.SingleProcessHBaseCluster; 042import org.apache.hadoop.hbase.StartTestingClusterOption; 043import org.apache.hadoop.hbase.TableName; 044import org.apache.hadoop.hbase.UnknownRegionException; 045import org.apache.hadoop.hbase.client.Admin; 046import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; 047import org.apache.hadoop.hbase.client.DoNotRetryRegionException; 048import org.apache.hadoop.hbase.client.Put; 049import org.apache.hadoop.hbase.client.RegionInfo; 050import org.apache.hadoop.hbase.client.RegionReplicaUtil; 051import org.apache.hadoop.hbase.client.Result; 052import org.apache.hadoop.hbase.client.ResultScanner; 053import org.apache.hadoop.hbase.client.Scan; 054import org.apache.hadoop.hbase.client.Table; 055import org.apache.hadoop.hbase.client.TableDescriptor; 056import org.apache.hadoop.hbase.exceptions.MergeRegionException; 057import org.apache.hadoop.hbase.master.HMaster; 058import org.apache.hadoop.hbase.master.MasterRpcServices; 059import org.apache.hadoop.hbase.master.RegionState; 060import org.apache.hadoop.hbase.master.assignment.AssignmentManager; 061import org.apache.hadoop.hbase.master.assignment.RegionStates; 062import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility; 063import org.apache.hadoop.hbase.testclassification.LargeTests; 064import org.apache.hadoop.hbase.testclassification.RegionServerTests; 065import org.apache.hadoop.hbase.util.Bytes; 066import org.apache.hadoop.hbase.util.CommonFSUtils; 067import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 068import org.apache.hadoop.hbase.util.FutureUtils; 069import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread; 070import org.apache.hadoop.hbase.util.Pair; 071import org.apache.hadoop.hbase.util.PairOfSameType; 072import org.apache.hadoop.hbase.util.Threads; 073import org.apache.hadoop.util.StringUtils; 074import org.apache.zookeeper.KeeperException; 075import org.junit.AfterClass; 076import org.junit.BeforeClass; 077import org.junit.ClassRule; 078import org.junit.Rule; 079import org.junit.Test; 080import org.junit.experimental.categories.Category; 081import org.junit.rules.TestName; 082import org.slf4j.Logger; 083import org.slf4j.LoggerFactory; 084 085import org.apache.hbase.thirdparty.com.google.common.base.Joiner; 086import org.apache.hbase.thirdparty.com.google.protobuf.RpcController; 087import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; 088 089import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionStateTransition.TransitionCode; 090import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionRequest; 091import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionResponse; 092 093@Category({ RegionServerTests.class, LargeTests.class }) 094public class TestRegionMergeTransactionOnCluster { 095 096 @ClassRule 097 public static final HBaseClassTestRule CLASS_RULE = 098 HBaseClassTestRule.forClass(TestRegionMergeTransactionOnCluster.class); 099 100 private static final Logger LOG = 101 LoggerFactory.getLogger(TestRegionMergeTransactionOnCluster.class); 102 103 @Rule 104 public TestName name = new TestName(); 105 106 private static final int NB_SERVERS = 3; 107 108 private static final byte[] FAMILYNAME = Bytes.toBytes("fam"); 109 private static final byte[] QUALIFIER = Bytes.toBytes("q"); 110 111 private static byte[] ROW = Bytes.toBytes("testRow"); 112 private static final int INITIAL_REGION_NUM = 10; 113 private static final int ROWSIZE = 200; 114 private static byte[][] ROWS = makeN(ROW, ROWSIZE); 115 116 private static int waitTime = 60 * 1000; 117 118 static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 119 120 private static HMaster MASTER; 121 private static Admin ADMIN; 122 123 @BeforeClass 124 public static void beforeAllTests() throws Exception { 125 // Start a cluster 126 StartTestingClusterOption option = StartTestingClusterOption.builder() 127 .masterClass(MyMaster.class).numRegionServers(NB_SERVERS).numDataNodes(NB_SERVERS).build(); 128 TEST_UTIL.startMiniCluster(option); 129 SingleProcessHBaseCluster cluster = TEST_UTIL.getHBaseCluster(); 130 MASTER = cluster.getMaster(); 131 MASTER.balanceSwitch(false); 132 ADMIN = TEST_UTIL.getConnection().getAdmin(); 133 } 134 135 @AfterClass 136 public static void afterAllTests() throws Exception { 137 TEST_UTIL.shutdownMiniCluster(); 138 if (ADMIN != null) { 139 ADMIN.close(); 140 } 141 } 142 143 @Test 144 public void testWholesomeMerge() throws Exception { 145 LOG.info("Starting " + name.getMethodName()); 146 final TableName tableName = TableName.valueOf(name.getMethodName()); 147 148 try { 149 // Create table and load data. 150 Table table = createTableAndLoadData(MASTER, tableName); 151 // Merge 1st and 2nd region 152 mergeRegionsAndVerifyRegionNum(MASTER, tableName, 0, 1, INITIAL_REGION_NUM - 1); 153 154 // Merge 2nd and 3th region 155 PairOfSameType<RegionInfo> mergedRegions = 156 mergeRegionsAndVerifyRegionNum(MASTER, tableName, 1, 2, INITIAL_REGION_NUM - 2); 157 158 verifyRowCount(table, ROWSIZE); 159 160 // Randomly choose one of the two merged regions 161 RegionInfo hri = ThreadLocalRandom.current().nextBoolean() 162 ? mergedRegions.getFirst() 163 : mergedRegions.getSecond(); 164 SingleProcessHBaseCluster cluster = TEST_UTIL.getHBaseCluster(); 165 AssignmentManager am = cluster.getMaster().getAssignmentManager(); 166 RegionStates regionStates = am.getRegionStates(); 167 168 // We should not be able to assign it again 169 am.assign(hri); 170 assertFalse("Merged region can't be assigned", regionStates.isRegionInTransition(hri)); 171 172 // We should not be able to unassign it either 173 am.unassign(hri); 174 assertFalse("Merged region can't be unassigned", regionStates.isRegionInTransition(hri)); 175 176 table.close(); 177 } finally { 178 TEST_UTIL.deleteTable(tableName); 179 } 180 } 181 182 /** 183 * Not really restarting the master. Simulate it by clear of new region state since it is not 184 * persisted, will be lost after master restarts. 185 */ 186 @Test 187 public void testMergeAndRestartingMaster() throws Exception { 188 final TableName tableName = TableName.valueOf(name.getMethodName()); 189 190 try { 191 // Create table and load data. 192 Table table = createTableAndLoadData(MASTER, tableName); 193 194 try { 195 MyMasterRpcServices.enabled.set(true); 196 197 // Merge 1st and 2nd region 198 mergeRegionsAndVerifyRegionNum(MASTER, tableName, 0, 1, INITIAL_REGION_NUM - 1); 199 } finally { 200 MyMasterRpcServices.enabled.set(false); 201 } 202 203 table.close(); 204 } finally { 205 TEST_UTIL.deleteTable(tableName); 206 } 207 } 208 209 @Test 210 public void testCleanMergeReference() throws Exception { 211 LOG.info("Starting " + name.getMethodName()); 212 ADMIN.catalogJanitorSwitch(false); 213 final TableName tableName = TableName.valueOf(name.getMethodName()); 214 try { 215 // Create table and load data. 216 Table table = createTableAndLoadData(MASTER, tableName); 217 // Merge 1st and 2nd region 218 mergeRegionsAndVerifyRegionNum(MASTER, tableName, 0, 1, INITIAL_REGION_NUM - 1); 219 verifyRowCount(table, ROWSIZE); 220 table.close(); 221 222 List<Pair<RegionInfo, ServerName>> tableRegions = 223 MetaTableAccessor.getTableRegionsAndLocations(MASTER.getConnection(), tableName); 224 RegionInfo mergedRegionInfo = tableRegions.get(0).getFirst(); 225 TableDescriptor tableDescriptor = MASTER.getTableDescriptors().get(tableName); 226 Result mergedRegionResult = 227 MetaTableAccessor.getRegionResult(MASTER.getConnection(), mergedRegionInfo); 228 229 // contains merge reference in META 230 assertTrue(CatalogFamilyFormat.hasMergeRegions(mergedRegionResult.rawCells())); 231 232 // merging regions' directory are in the file system all the same 233 List<RegionInfo> p = CatalogFamilyFormat.getMergeRegions(mergedRegionResult.rawCells()); 234 RegionInfo regionA = p.get(0); 235 RegionInfo regionB = p.get(1); 236 FileSystem fs = MASTER.getMasterFileSystem().getFileSystem(); 237 Path rootDir = MASTER.getMasterFileSystem().getRootDir(); 238 239 Path tabledir = CommonFSUtils.getTableDir(rootDir, mergedRegionInfo.getTable()); 240 Path regionAdir = new Path(tabledir, regionA.getEncodedName()); 241 Path regionBdir = new Path(tabledir, regionB.getEncodedName()); 242 assertTrue(fs.exists(regionAdir)); 243 assertTrue(fs.exists(regionBdir)); 244 245 ColumnFamilyDescriptor[] columnFamilies = tableDescriptor.getColumnFamilies(); 246 HRegionFileSystem hrfs = 247 new HRegionFileSystem(TEST_UTIL.getConfiguration(), fs, tabledir, mergedRegionInfo); 248 int count = 0; 249 for (ColumnFamilyDescriptor colFamily : columnFamilies) { 250 count += hrfs.getStoreFiles(colFamily.getNameAsString()).size(); 251 } 252 ADMIN.compactRegion(mergedRegionInfo.getRegionName()); 253 // clean up the merged region store files 254 // wait until merged region have reference file 255 long timeout = EnvironmentEdgeManager.currentTime() + waitTime; 256 int newcount = 0; 257 while (EnvironmentEdgeManager.currentTime() < timeout) { 258 for (ColumnFamilyDescriptor colFamily : columnFamilies) { 259 newcount += hrfs.getStoreFiles(colFamily.getNameAsString()).size(); 260 } 261 if (newcount > count) { 262 break; 263 } 264 Thread.sleep(50); 265 } 266 assertTrue(newcount > count); 267 List<RegionServerThread> regionServerThreads = 268 TEST_UTIL.getHBaseCluster().getRegionServerThreads(); 269 for (RegionServerThread rs : regionServerThreads) { 270 CompactedHFilesDischarger cleaner = 271 new CompactedHFilesDischarger(100, null, rs.getRegionServer(), false); 272 cleaner.chore(); 273 Thread.sleep(1000); 274 } 275 while (EnvironmentEdgeManager.currentTime() < timeout) { 276 int newcount1 = 0; 277 for (ColumnFamilyDescriptor colFamily : columnFamilies) { 278 newcount1 += hrfs.getStoreFiles(colFamily.getNameAsString()).size(); 279 } 280 if (newcount1 <= 1) { 281 break; 282 } 283 Thread.sleep(50); 284 } 285 // run CatalogJanitor to clean merge references in hbase:meta and archive the 286 // files of merging regions 287 int cleaned = 0; 288 while (cleaned == 0) { 289 cleaned = ADMIN.runCatalogJanitor(); 290 LOG.debug("catalog janitor returned " + cleaned); 291 Thread.sleep(50); 292 // Cleanup is async so wait till all procedures are done running. 293 ProcedureTestingUtility.waitNoProcedureRunning( 294 TEST_UTIL.getMiniHBaseCluster().getMaster().getMasterProcedureExecutor()); 295 } 296 // We used to check for existence of region in fs but sometimes the region dir was 297 // cleaned up by the time we got here making the test sometimes flakey. 298 assertTrue(cleaned > 0); 299 300 // Wait around a bit to give stuff a chance to complete. 301 while (true) { 302 mergedRegionResult = 303 MetaTableAccessor.getRegionResult(TEST_UTIL.getConnection(), mergedRegionInfo); 304 if (CatalogFamilyFormat.hasMergeRegions(mergedRegionResult.rawCells())) { 305 LOG.info("Waiting on cleanup of merge columns {}", 306 Arrays.asList(mergedRegionResult.rawCells()).stream().map(c -> c.toString()) 307 .collect(Collectors.joining(","))); 308 Threads.sleep(50); 309 } else { 310 break; 311 } 312 } 313 assertFalse(CatalogFamilyFormat.hasMergeRegions(mergedRegionResult.rawCells())); 314 } finally { 315 ADMIN.catalogJanitorSwitch(true); 316 TEST_UTIL.deleteTable(tableName); 317 } 318 } 319 320 /** 321 * This test tests 1, merging region not online; 2, merging same two regions; 3, merging unknown 322 * regions. They are in one test case so that we don't have to create many tables, and these tests 323 * are simple. 324 */ 325 @Test 326 public void testMerge() throws Exception { 327 LOG.info("Starting " + name.getMethodName()); 328 final TableName tableName = TableName.valueOf(name.getMethodName()); 329 final Admin admin = TEST_UTIL.getAdmin(); 330 331 try { 332 // Create table and load data. 333 Table table = createTableAndLoadData(MASTER, tableName); 334 AssignmentManager am = MASTER.getAssignmentManager(); 335 List<RegionInfo> regions = am.getRegionStates().getRegionsOfTable(tableName); 336 // Fake offline one region 337 RegionInfo a = regions.get(0); 338 RegionInfo b = regions.get(1); 339 am.unassign(b); 340 am.offlineRegion(b); 341 try { 342 // Merge offline region. Region a is offline here 343 FutureUtils.get( 344 admin.mergeRegionsAsync(a.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), false)); 345 fail("Offline regions should not be able to merge"); 346 } catch (DoNotRetryRegionException ie) { 347 System.out.println(ie); 348 assertTrue(ie instanceof MergeRegionException); 349 } 350 351 try { 352 // Merge the same region: b and b. 353 FutureUtils 354 .get(admin.mergeRegionsAsync(b.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), true)); 355 fail("A region should not be able to merge with itself, even forcfully"); 356 } catch (IOException ie) { 357 assertTrue("Exception should mention regions not online", 358 StringUtils.stringifyException(ie).contains("region to itself") 359 && ie instanceof MergeRegionException); 360 } 361 362 try { 363 // Merge unknown regions 364 FutureUtils.get(admin.mergeRegionsAsync(Bytes.toBytes("-f1"), Bytes.toBytes("-f2"), true)); 365 fail("Unknown region could not be merged"); 366 } catch (IOException ie) { 367 assertTrue("UnknownRegionException should be thrown", ie instanceof UnknownRegionException); 368 } 369 table.close(); 370 } finally { 371 TEST_UTIL.deleteTable(tableName); 372 } 373 } 374 375 @Test 376 public void testMergeWithReplicas() throws Exception { 377 final TableName tableName = TableName.valueOf(name.getMethodName()); 378 try { 379 // Create table and load data. 380 Table table = createTableAndLoadData(MASTER, tableName, 5, 2); 381 List<Pair<RegionInfo, ServerName>> initialRegionToServers = 382 MetaTableAccessor.getTableRegionsAndLocations(TEST_UTIL.getConnection(), tableName); 383 // Merge 1st and 2nd region 384 PairOfSameType<RegionInfo> mergedRegions = 385 mergeRegionsAndVerifyRegionNum(MASTER, tableName, 0, 2, 5 * 2 - 2); 386 List<Pair<RegionInfo, ServerName>> currentRegionToServers = 387 MetaTableAccessor.getTableRegionsAndLocations(TEST_UTIL.getConnection(), tableName); 388 List<RegionInfo> initialRegions = new ArrayList<>(); 389 for (Pair<RegionInfo, ServerName> p : initialRegionToServers) { 390 initialRegions.add(p.getFirst()); 391 } 392 List<RegionInfo> currentRegions = new ArrayList<>(); 393 for (Pair<RegionInfo, ServerName> p : currentRegionToServers) { 394 currentRegions.add(p.getFirst()); 395 } 396 // this is the first region 397 assertTrue(initialRegions.contains(mergedRegions.getFirst())); 398 // this is the replica of the first region 399 assertTrue(initialRegions 400 .contains(RegionReplicaUtil.getRegionInfoForReplica(mergedRegions.getFirst(), 1))); 401 // this is the second region 402 assertTrue(initialRegions.contains(mergedRegions.getSecond())); 403 // this is the replica of the second region 404 assertTrue(initialRegions 405 .contains(RegionReplicaUtil.getRegionInfoForReplica(mergedRegions.getSecond(), 1))); 406 // this is the new region 407 assertTrue(!initialRegions.contains(currentRegions.get(0))); 408 // replica of the new region 409 assertTrue(!initialRegions 410 .contains(RegionReplicaUtil.getRegionInfoForReplica(currentRegions.get(0), 1))); 411 // replica of the new region 412 assertTrue(currentRegions 413 .contains(RegionReplicaUtil.getRegionInfoForReplica(currentRegions.get(0), 1))); 414 // replica of the merged region 415 assertTrue(!currentRegions 416 .contains(RegionReplicaUtil.getRegionInfoForReplica(mergedRegions.getFirst(), 1))); 417 // replica of the merged region 418 assertTrue(!currentRegions 419 .contains(RegionReplicaUtil.getRegionInfoForReplica(mergedRegions.getSecond(), 1))); 420 table.close(); 421 } finally { 422 TEST_UTIL.deleteTable(tableName); 423 } 424 } 425 426 private PairOfSameType<RegionInfo> mergeRegionsAndVerifyRegionNum(HMaster master, 427 TableName tablename, int regionAnum, int regionBnum, int expectedRegionNum) throws Exception { 428 PairOfSameType<RegionInfo> mergedRegions = 429 requestMergeRegion(master, tablename, regionAnum, regionBnum); 430 waitAndVerifyRegionNum(master, tablename, expectedRegionNum); 431 return mergedRegions; 432 } 433 434 private PairOfSameType<RegionInfo> requestMergeRegion(HMaster master, TableName tablename, 435 int regionAnum, int regionBnum) throws Exception { 436 List<Pair<RegionInfo, ServerName>> tableRegions = 437 MetaTableAccessor.getTableRegionsAndLocations(TEST_UTIL.getConnection(), tablename); 438 RegionInfo regionA = tableRegions.get(regionAnum).getFirst(); 439 RegionInfo regionB = tableRegions.get(regionBnum).getFirst(); 440 ADMIN.mergeRegionsAsync(regionA.getEncodedNameAsBytes(), regionB.getEncodedNameAsBytes(), 441 false); 442 return new PairOfSameType<>(regionA, regionB); 443 } 444 445 private void waitAndVerifyRegionNum(HMaster master, TableName tablename, int expectedRegionNum) 446 throws Exception { 447 List<Pair<RegionInfo, ServerName>> tableRegionsInMeta; 448 List<RegionInfo> tableRegionsInMaster; 449 long timeout = EnvironmentEdgeManager.currentTime() + waitTime; 450 while (EnvironmentEdgeManager.currentTime() < timeout) { 451 tableRegionsInMeta = 452 MetaTableAccessor.getTableRegionsAndLocations(TEST_UTIL.getConnection(), tablename); 453 tableRegionsInMaster = 454 master.getAssignmentManager().getRegionStates().getRegionsOfTable(tablename); 455 LOG.info(Objects.toString(tableRegionsInMaster)); 456 LOG.info(Objects.toString(tableRegionsInMeta)); 457 int tableRegionsInMetaSize = tableRegionsInMeta.size(); 458 int tableRegionsInMasterSize = tableRegionsInMaster.size(); 459 if ( 460 tableRegionsInMetaSize == expectedRegionNum && tableRegionsInMasterSize == expectedRegionNum 461 ) { 462 break; 463 } 464 Thread.sleep(250); 465 } 466 467 tableRegionsInMeta = 468 MetaTableAccessor.getTableRegionsAndLocations(TEST_UTIL.getConnection(), tablename); 469 LOG.info("Regions after merge:" + Joiner.on(',').join(tableRegionsInMeta)); 470 assertEquals(expectedRegionNum, tableRegionsInMeta.size()); 471 } 472 473 private Table createTableAndLoadData(HMaster master, TableName tablename) throws Exception { 474 return createTableAndLoadData(master, tablename, INITIAL_REGION_NUM, 1); 475 } 476 477 private Table createTableAndLoadData(HMaster master, TableName tablename, int numRegions, 478 int replication) throws Exception { 479 assertTrue("ROWSIZE must > numregions:" + numRegions, ROWSIZE > numRegions); 480 byte[][] splitRows = new byte[numRegions - 1][]; 481 for (int i = 0; i < splitRows.length; i++) { 482 splitRows[i] = ROWS[(i + 1) * ROWSIZE / numRegions]; 483 } 484 485 Table table = TEST_UTIL.createTable(tablename, FAMILYNAME, splitRows); 486 LOG.info("Created " + table.getName()); 487 if (replication > 1) { 488 HBaseTestingUtil.setReplicas(ADMIN, tablename, replication); 489 LOG.info("Set replication of " + replication + " on " + table.getName()); 490 } 491 loadData(table); 492 LOG.info("Loaded " + table.getName()); 493 verifyRowCount(table, ROWSIZE); 494 LOG.info("Verified " + table.getName()); 495 496 List<Pair<RegionInfo, ServerName>> tableRegions; 497 TEST_UTIL.waitUntilAllRegionsAssigned(tablename); 498 LOG.info("All regions assigned for table - " + table.getName()); 499 tableRegions = 500 MetaTableAccessor.getTableRegionsAndLocations(TEST_UTIL.getConnection(), tablename); 501 assertEquals("Wrong number of regions in table " + tablename, numRegions * replication, 502 tableRegions.size()); 503 LOG.info(tableRegions.size() + "Regions after load: " + Joiner.on(',').join(tableRegions)); 504 assertEquals(numRegions * replication, tableRegions.size()); 505 return table; 506 } 507 508 private static byte[][] makeN(byte[] base, int n) { 509 byte[][] ret = new byte[n][]; 510 for (int i = 0; i < n; i++) { 511 ret[i] = Bytes.add(base, Bytes.toBytes(String.format("%04d", i))); 512 } 513 return ret; 514 } 515 516 private void loadData(Table table) throws IOException { 517 for (int i = 0; i < ROWSIZE; i++) { 518 Put put = new Put(ROWS[i]); 519 put.addColumn(FAMILYNAME, QUALIFIER, Bytes.toBytes(i)); 520 table.put(put); 521 } 522 } 523 524 private void verifyRowCount(Table table, int expectedRegionNum) throws IOException { 525 ResultScanner scanner = table.getScanner(new Scan()); 526 int rowCount = 0; 527 while (scanner.next() != null) { 528 rowCount++; 529 } 530 assertEquals(expectedRegionNum, rowCount); 531 scanner.close(); 532 } 533 534 // Make it public so that JVMClusterUtil can access it. 535 public static class MyMaster extends HMaster { 536 public MyMaster(Configuration conf) throws IOException, KeeperException, InterruptedException { 537 super(conf); 538 } 539 540 @Override 541 protected MasterRpcServices createRpcServices() throws IOException { 542 return new MyMasterRpcServices(this); 543 } 544 } 545 546 static class MyMasterRpcServices extends MasterRpcServices { 547 static AtomicBoolean enabled = new AtomicBoolean(false); 548 549 private HMaster myMaster; 550 551 public MyMasterRpcServices(HMaster master) throws IOException { 552 super(master); 553 myMaster = master; 554 } 555 556 @Override 557 public ReportRegionStateTransitionResponse reportRegionStateTransition(RpcController c, 558 ReportRegionStateTransitionRequest req) throws ServiceException { 559 ReportRegionStateTransitionResponse resp = super.reportRegionStateTransition(c, req); 560 if ( 561 enabled.get() && req.getTransition(0).getTransitionCode() == TransitionCode.READY_TO_MERGE 562 && !resp.hasErrorMessage() 563 ) { 564 RegionStates regionStates = myMaster.getAssignmentManager().getRegionStates(); 565 for (RegionState regionState : regionStates.getRegionsStateInTransition()) { 566 // Find the merging_new region and remove it 567 if (regionState.isMergingNew()) { 568 regionStates.deleteRegion(regionState.getRegion()); 569 } 570 } 571 } 572 return resp; 573 } 574 } 575}