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; 019 020import java.io.File; 021import java.io.IOException; 022import java.io.OutputStream; 023import java.io.UncheckedIOException; 024import java.lang.reflect.Field; 025import java.lang.reflect.Modifier; 026import java.net.BindException; 027import java.net.DatagramSocket; 028import java.net.InetAddress; 029import java.net.ServerSocket; 030import java.net.Socket; 031import java.net.UnknownHostException; 032import java.nio.charset.StandardCharsets; 033import java.security.MessageDigest; 034import java.util.ArrayList; 035import java.util.Arrays; 036import java.util.Collection; 037import java.util.Collections; 038import java.util.HashSet; 039import java.util.Iterator; 040import java.util.List; 041import java.util.Map; 042import java.util.NavigableSet; 043import java.util.Properties; 044import java.util.Random; 045import java.util.Set; 046import java.util.TreeSet; 047import java.util.concurrent.ThreadLocalRandom; 048import java.util.concurrent.TimeUnit; 049import java.util.concurrent.atomic.AtomicReference; 050import java.util.function.BooleanSupplier; 051import org.apache.commons.io.FileUtils; 052import org.apache.commons.lang3.RandomStringUtils; 053import org.apache.hadoop.conf.Configuration; 054import org.apache.hadoop.fs.FileSystem; 055import org.apache.hadoop.fs.Path; 056import org.apache.hadoop.hbase.Waiter.ExplainingPredicate; 057import org.apache.hadoop.hbase.Waiter.Predicate; 058import org.apache.hadoop.hbase.client.Admin; 059import org.apache.hadoop.hbase.client.AsyncClusterConnection; 060import org.apache.hadoop.hbase.client.BufferedMutator; 061import org.apache.hadoop.hbase.client.ClusterConnectionFactory; 062import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; 063import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 064import org.apache.hadoop.hbase.client.Connection; 065import org.apache.hadoop.hbase.client.ConnectionFactory; 066import org.apache.hadoop.hbase.client.Consistency; 067import org.apache.hadoop.hbase.client.Delete; 068import org.apache.hadoop.hbase.client.Durability; 069import org.apache.hadoop.hbase.client.Get; 070import org.apache.hadoop.hbase.client.Hbck; 071import org.apache.hadoop.hbase.client.MasterRegistry; 072import org.apache.hadoop.hbase.client.Put; 073import org.apache.hadoop.hbase.client.RegionInfo; 074import org.apache.hadoop.hbase.client.RegionInfoBuilder; 075import org.apache.hadoop.hbase.client.RegionLocator; 076import org.apache.hadoop.hbase.client.Result; 077import org.apache.hadoop.hbase.client.ResultScanner; 078import org.apache.hadoop.hbase.client.Scan; 079import org.apache.hadoop.hbase.client.Scan.ReadType; 080import org.apache.hadoop.hbase.client.Table; 081import org.apache.hadoop.hbase.client.TableDescriptor; 082import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 083import org.apache.hadoop.hbase.client.TableState; 084import org.apache.hadoop.hbase.fs.HFileSystem; 085import org.apache.hadoop.hbase.io.compress.Compression; 086import org.apache.hadoop.hbase.io.compress.Compression.Algorithm; 087import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding; 088import org.apache.hadoop.hbase.io.hfile.BlockCache; 089import org.apache.hadoop.hbase.io.hfile.ChecksumUtil; 090import org.apache.hadoop.hbase.io.hfile.HFile; 091import org.apache.hadoop.hbase.ipc.RpcServerInterface; 092import org.apache.hadoop.hbase.logging.Log4jUtils; 093import org.apache.hadoop.hbase.mapreduce.MapreduceTestingShim; 094import org.apache.hadoop.hbase.master.HMaster; 095import org.apache.hadoop.hbase.master.RegionState; 096import org.apache.hadoop.hbase.master.ServerManager; 097import org.apache.hadoop.hbase.master.assignment.AssignmentManager; 098import org.apache.hadoop.hbase.master.assignment.AssignmentTestingUtil; 099import org.apache.hadoop.hbase.master.assignment.RegionStateStore; 100import org.apache.hadoop.hbase.master.assignment.RegionStates; 101import org.apache.hadoop.hbase.mob.MobFileCache; 102import org.apache.hadoop.hbase.regionserver.BloomType; 103import org.apache.hadoop.hbase.regionserver.ChunkCreator; 104import org.apache.hadoop.hbase.regionserver.HRegion; 105import org.apache.hadoop.hbase.regionserver.HRegionServer; 106import org.apache.hadoop.hbase.regionserver.HStore; 107import org.apache.hadoop.hbase.regionserver.InternalScanner; 108import org.apache.hadoop.hbase.regionserver.MemStoreLAB; 109import org.apache.hadoop.hbase.regionserver.Region; 110import org.apache.hadoop.hbase.regionserver.RegionScanner; 111import org.apache.hadoop.hbase.regionserver.RegionServerServices; 112import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException; 113import org.apache.hadoop.hbase.security.HBaseKerberosUtils; 114import org.apache.hadoop.hbase.security.User; 115import org.apache.hadoop.hbase.security.UserProvider; 116import org.apache.hadoop.hbase.security.visibility.VisibilityLabelsCache; 117import org.apache.hadoop.hbase.util.Bytes; 118import org.apache.hadoop.hbase.util.CommonFSUtils; 119import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 120import org.apache.hadoop.hbase.util.FSUtils; 121import org.apache.hadoop.hbase.util.JVMClusterUtil; 122import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread; 123import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread; 124import org.apache.hadoop.hbase.util.Pair; 125import org.apache.hadoop.hbase.util.ReflectionUtils; 126import org.apache.hadoop.hbase.util.RegionSplitter; 127import org.apache.hadoop.hbase.util.RegionSplitter.SplitAlgorithm; 128import org.apache.hadoop.hbase.util.RetryCounter; 129import org.apache.hadoop.hbase.util.Threads; 130import org.apache.hadoop.hbase.wal.WAL; 131import org.apache.hadoop.hbase.wal.WALFactory; 132import org.apache.hadoop.hbase.zookeeper.EmptyWatcher; 133import org.apache.hadoop.hbase.zookeeper.ZKConfig; 134import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 135import org.apache.hadoop.hdfs.DFSClient; 136import org.apache.hadoop.hdfs.DFSConfigKeys; 137import org.apache.hadoop.hdfs.DistributedFileSystem; 138import org.apache.hadoop.hdfs.MiniDFSCluster; 139import org.apache.hadoop.hdfs.server.datanode.DataNode; 140import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi; 141import org.apache.hadoop.hdfs.server.namenode.EditLogFileOutputStream; 142import org.apache.hadoop.mapred.JobConf; 143import org.apache.hadoop.mapred.MiniMRCluster; 144import org.apache.hadoop.mapred.TaskLog; 145import org.apache.hadoop.metrics2.impl.JmxCacheBuster; 146import org.apache.hadoop.minikdc.MiniKdc; 147import org.apache.yetus.audience.InterfaceAudience; 148import org.apache.zookeeper.WatchedEvent; 149import org.apache.zookeeper.ZooKeeper; 150import org.apache.zookeeper.ZooKeeper.States; 151 152import org.apache.hbase.thirdparty.com.google.common.io.Closeables; 153 154import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 155 156/** 157 * Facility for testing HBase. Replacement for old HBaseTestCase and HBaseClusterTestCase 158 * functionality. Create an instance and keep it around testing HBase. This class is meant to be 159 * your one-stop shop for anything you might need testing. Manages one cluster at a time only. 160 * Managed cluster can be an in-process {@link MiniHBaseCluster}, or a deployed cluster of type 161 * {@code DistributedHBaseCluster}. Not all methods work with the real cluster. Depends on log4j 162 * being on classpath and hbase-site.xml for logging and test-run configuration. It does not set 163 * logging levels. In the configuration properties, default values for master-info-port and 164 * region-server-port are overridden such that a random port will be assigned (thus avoiding port 165 * contention if another local HBase instance is already running). 166 * <p> 167 * To preserve test data directories, pass the system property "hbase.testing.preserve.testdir" 168 * setting it to true. 169 * @deprecated since 3.0.0, will be removed in 4.0.0. Use 170 * {@link org.apache.hadoop.hbase.testing.TestingHBaseCluster} instead. 171 */ 172@InterfaceAudience.Public 173@Deprecated 174public class HBaseTestingUtility extends HBaseZKTestingUtility { 175 176 public static final String REGIONS_PER_SERVER_KEY = "hbase.test.regions-per-server"; 177 /** 178 * The default number of regions per regionserver when creating a pre-split table. 179 */ 180 public static final int DEFAULT_REGIONS_PER_SERVER = 3; 181 182 private MiniDFSCluster dfsCluster = null; 183 private FsDatasetAsyncDiskServiceFixer dfsClusterFixer = null; 184 185 private volatile HBaseCluster hbaseCluster = null; 186 private MiniMRCluster mrCluster = null; 187 188 /** If there is a mini cluster running for this testing utility instance. */ 189 private volatile boolean miniClusterRunning; 190 191 private String hadoopLogDir; 192 193 /** 194 * Directory on test filesystem where we put the data for this instance of HBaseTestingUtility 195 */ 196 private Path dataTestDirOnTestFS = null; 197 198 private final AtomicReference<AsyncClusterConnection> asyncConnection = new AtomicReference<>(); 199 200 /** Filesystem URI used for map-reduce mini-cluster setup */ 201 private static String FS_URI; 202 203 /** This is for unit tests parameterized with a single boolean. */ 204 public static final List<Object[]> MEMSTORETS_TAGS_PARAMETRIZED = memStoreTSAndTagsCombination(); 205 206 static { 207 // JmxCacheBuster may cause dead lock in test environment. As on master side, the table/region 208 // related metrics updating will finally lead to a meta access, so if meta is not online yet, we 209 // will block when updating while holding the metrics lock. But when we assign meta, there are 210 // bunch of places where we need to register a new metrics thus need to get the metrics lock, 211 // and then lead to a dead lock and cause the test to hang forever. 212 // The code is in hadoop so there is no easy way for us to fix, so here we just stop 213 // JmxCacheBuster to stabilize our tests first. See HBASE-30118 for more details and future 214 // plans. 215 JmxCacheBuster.stop(); 216 } 217 218 /** 219 * Checks to see if a specific port is available. 220 * @param port the port number to check for availability 221 * @return <tt>true</tt> if the port is available, or <tt>false</tt> if not 222 */ 223 public static boolean available(int port) { 224 ServerSocket ss = null; 225 DatagramSocket ds = null; 226 try { 227 ss = new ServerSocket(port); 228 ss.setReuseAddress(true); 229 ds = new DatagramSocket(port); 230 ds.setReuseAddress(true); 231 return true; 232 } catch (IOException e) { 233 // Do nothing 234 } finally { 235 if (ds != null) { 236 ds.close(); 237 } 238 239 if (ss != null) { 240 try { 241 ss.close(); 242 } catch (IOException e) { 243 /* should not be thrown */ 244 } 245 } 246 } 247 248 return false; 249 } 250 251 /** 252 * Create all combinations of Bloom filters and compression algorithms for testing. 253 */ 254 private static List<Object[]> bloomAndCompressionCombinations() { 255 List<Object[]> configurations = new ArrayList<>(); 256 for (Compression.Algorithm comprAlgo : HBaseCommonTestingUtility.COMPRESSION_ALGORITHMS) { 257 for (BloomType bloomType : BloomType.values()) { 258 configurations.add(new Object[] { comprAlgo, bloomType }); 259 } 260 } 261 return Collections.unmodifiableList(configurations); 262 } 263 264 /** 265 * Create combination of memstoreTS and tags 266 */ 267 private static List<Object[]> memStoreTSAndTagsCombination() { 268 List<Object[]> configurations = new ArrayList<>(); 269 configurations.add(new Object[] { false, false }); 270 configurations.add(new Object[] { false, true }); 271 configurations.add(new Object[] { true, false }); 272 configurations.add(new Object[] { true, true }); 273 return Collections.unmodifiableList(configurations); 274 } 275 276 public static List<Object[]> memStoreTSTagsAndOffheapCombination() { 277 List<Object[]> configurations = new ArrayList<>(); 278 configurations.add(new Object[] { false, false, true }); 279 configurations.add(new Object[] { false, false, false }); 280 configurations.add(new Object[] { false, true, true }); 281 configurations.add(new Object[] { false, true, false }); 282 configurations.add(new Object[] { true, false, true }); 283 configurations.add(new Object[] { true, false, false }); 284 configurations.add(new Object[] { true, true, true }); 285 configurations.add(new Object[] { true, true, false }); 286 return Collections.unmodifiableList(configurations); 287 } 288 289 public static final Collection<Object[]> BLOOM_AND_COMPRESSION_COMBINATIONS = 290 bloomAndCompressionCombinations(); 291 292 /** 293 * <p> 294 * Create an HBaseTestingUtility using a default configuration. 295 * <p> 296 * Initially, all tmp files are written to a local test data directory. Once 297 * {@link #startMiniDFSCluster} is called, either directly or via {@link #startMiniCluster()}, tmp 298 * data will be written to the DFS directory instead. 299 */ 300 public HBaseTestingUtility() { 301 this(HBaseConfiguration.create()); 302 } 303 304 /** 305 * <p> 306 * Create an HBaseTestingUtility using a given configuration. 307 * <p> 308 * Initially, all tmp files are written to a local test data directory. Once 309 * {@link #startMiniDFSCluster} is called, either directly or via {@link #startMiniCluster()}, tmp 310 * data will be written to the DFS directory instead. 311 * @param conf The configuration to use for further operations 312 */ 313 public HBaseTestingUtility(Configuration conf) { 314 super(conf); 315 316 // a hbase checksum verification failure will cause unit tests to fail 317 ChecksumUtil.generateExceptionForChecksumFailureForTest(true); 318 319 // Save this for when setting default file:// breaks things 320 if (this.conf.get("fs.defaultFS") != null) { 321 this.conf.set("original.defaultFS", this.conf.get("fs.defaultFS")); 322 } 323 if (this.conf.get(HConstants.HBASE_DIR) != null) { 324 this.conf.set("original.hbase.dir", this.conf.get(HConstants.HBASE_DIR)); 325 } 326 // Every cluster is a local cluster until we start DFS 327 // Note that conf could be null, but this.conf will not be 328 String dataTestDir = getDataTestDir().toString(); 329 this.conf.set("fs.defaultFS", "file:///"); 330 this.conf.set(HConstants.HBASE_DIR, "file://" + dataTestDir); 331 LOG.debug("Setting {} to {}", HConstants.HBASE_DIR, dataTestDir); 332 this.conf.setBoolean(CommonFSUtils.UNSAFE_STREAM_CAPABILITY_ENFORCE, false); 333 // If the value for random ports isn't set set it to true, thus making 334 // tests opt-out for random port assignment 335 this.conf.setBoolean(LocalHBaseCluster.ASSIGN_RANDOM_PORTS, 336 this.conf.getBoolean(LocalHBaseCluster.ASSIGN_RANDOM_PORTS, true)); 337 } 338 339 /** 340 * Close both the region {@code r} and it's underlying WAL. For use in tests. 341 */ 342 public static void closeRegionAndWAL(final Region r) throws IOException { 343 closeRegionAndWAL((HRegion) r); 344 } 345 346 /** 347 * Close both the HRegion {@code r} and it's underlying WAL. For use in tests. 348 */ 349 public static void closeRegionAndWAL(final HRegion r) throws IOException { 350 if (r == null) return; 351 r.close(); 352 if (r.getWAL() == null) return; 353 r.getWAL().close(); 354 } 355 356 /** 357 * Returns this classes's instance of {@link Configuration}. Be careful how you use the returned 358 * Configuration since {@link Connection} instances can be shared. The Map of Connections is keyed 359 * by the Configuration. If say, a Connection was being used against a cluster that had been 360 * shutdown, see {@link #shutdownMiniCluster()}, then the Connection will no longer be wholesome. 361 * Rather than use the return direct, its usually best to make a copy and use that. Do 362 * <code>Configuration c = new Configuration(INSTANCE.getConfiguration());</code> 363 * @return Instance of Configuration. 364 */ 365 @Override 366 public Configuration getConfiguration() { 367 return super.getConfiguration(); 368 } 369 370 public void setHBaseCluster(HBaseCluster hbaseCluster) { 371 this.hbaseCluster = hbaseCluster; 372 } 373 374 /** 375 * Home our data in a dir under {@link #DEFAULT_BASE_TEST_DIRECTORY}. Give it a random name so can 376 * have many concurrent tests running if we need to. Moding a System property is not the way to do 377 * concurrent instances -- another instance could grab the temporary value unintentionally -- but 378 * not anything can do about it at moment; single instance only is how the minidfscluster works. 379 * We also create the underlying directory names for hadoop.log.dir, mapreduce.cluster.local.dir 380 * and hadoop.tmp.dir, and set the values in the conf, and as a system property for hadoop.tmp.dir 381 * (We do not create them!). 382 * @return The calculated data test build directory, if newly-created. 383 */ 384 @Override 385 protected Path setupDataTestDir() { 386 Path testPath = super.setupDataTestDir(); 387 if (null == testPath) { 388 return null; 389 } 390 391 createSubDirAndSystemProperty("hadoop.log.dir", testPath, "hadoop-log-dir"); 392 393 // This is defaulted in core-default.xml to /tmp/hadoop-${user.name}, but 394 // we want our own value to ensure uniqueness on the same machine 395 createSubDirAndSystemProperty("hadoop.tmp.dir", testPath, "hadoop-tmp-dir"); 396 397 // Read and modified in org.apache.hadoop.mapred.MiniMRCluster 398 createSubDir("mapreduce.cluster.local.dir", testPath, "mapred-local-dir"); 399 return testPath; 400 } 401 402 private void createSubDirAndSystemProperty(String propertyName, Path parent, String subDirName) { 403 404 String sysValue = System.getProperty(propertyName); 405 406 if (sysValue != null) { 407 // There is already a value set. So we do nothing but hope 408 // that there will be no conflicts 409 LOG.info("System.getProperty(\"" + propertyName + "\") already set to: " + sysValue 410 + " so I do NOT create it in " + parent); 411 String confValue = conf.get(propertyName); 412 if (confValue != null && !confValue.endsWith(sysValue)) { 413 LOG.warn(propertyName + " property value differs in configuration and system: " 414 + "Configuration=" + confValue + " while System=" + sysValue 415 + " Erasing configuration value by system value."); 416 } 417 conf.set(propertyName, sysValue); 418 } else { 419 // Ok, it's not set, so we create it as a subdirectory 420 createSubDir(propertyName, parent, subDirName); 421 System.setProperty(propertyName, conf.get(propertyName)); 422 } 423 } 424 425 /** 426 * @return Where to write test data on the test filesystem; Returns working directory for the test 427 * filesystem by default 428 * @see #setupDataTestDirOnTestFS() 429 * @see #getTestFileSystem() 430 */ 431 private Path getBaseTestDirOnTestFS() throws IOException { 432 FileSystem fs = getTestFileSystem(); 433 return new Path(fs.getWorkingDirectory(), "test-data"); 434 } 435 436 /** 437 * Returns a Path in the test filesystem, obtained from {@link #getTestFileSystem()} to write 438 * temporary test data. Call this method after setting up the mini dfs cluster if the test relies 439 * on it. 440 * @return a unique path in the test filesystem 441 */ 442 public Path getDataTestDirOnTestFS() throws IOException { 443 if (dataTestDirOnTestFS == null) { 444 setupDataTestDirOnTestFS(); 445 } 446 447 return dataTestDirOnTestFS; 448 } 449 450 /** 451 * Returns a Path in the test filesystem, obtained from {@link #getTestFileSystem()} to write 452 * temporary test data. Call this method after setting up the mini dfs cluster if the test relies 453 * on it. 454 * @return a unique path in the test filesystem 455 * @param subdirName name of the subdir to create under the base test dir 456 */ 457 public Path getDataTestDirOnTestFS(final String subdirName) throws IOException { 458 return new Path(getDataTestDirOnTestFS(), subdirName); 459 } 460 461 /** 462 * Sets up a path in test filesystem to be used by tests. Creates a new directory if not already 463 * setup. 464 */ 465 private void setupDataTestDirOnTestFS() throws IOException { 466 if (dataTestDirOnTestFS != null) { 467 LOG.warn("Data test on test fs dir already setup in " + dataTestDirOnTestFS.toString()); 468 return; 469 } 470 dataTestDirOnTestFS = getNewDataTestDirOnTestFS(); 471 } 472 473 /** 474 * Sets up a new path in test filesystem to be used by tests. 475 */ 476 private Path getNewDataTestDirOnTestFS() throws IOException { 477 // The file system can be either local, mini dfs, or if the configuration 478 // is supplied externally, it can be an external cluster FS. If it is a local 479 // file system, the tests should use getBaseTestDir, otherwise, we can use 480 // the working directory, and create a unique sub dir there 481 FileSystem fs = getTestFileSystem(); 482 Path newDataTestDir; 483 String randomStr = getRandomUUID().toString(); 484 if (fs.getUri().getScheme().equals(FileSystem.getLocal(conf).getUri().getScheme())) { 485 newDataTestDir = new Path(getDataTestDir(), randomStr); 486 File dataTestDir = new File(newDataTestDir.toString()); 487 if (deleteOnExit()) dataTestDir.deleteOnExit(); 488 } else { 489 Path base = getBaseTestDirOnTestFS(); 490 newDataTestDir = new Path(base, randomStr); 491 if (deleteOnExit()) fs.deleteOnExit(newDataTestDir); 492 } 493 return newDataTestDir; 494 } 495 496 /** 497 * Cleans the test data directory on the test filesystem. 498 * @return True if we removed the test dirs 499 */ 500 public boolean cleanupDataTestDirOnTestFS() throws IOException { 501 boolean ret = getTestFileSystem().delete(dataTestDirOnTestFS, true); 502 if (ret) dataTestDirOnTestFS = null; 503 return ret; 504 } 505 506 /** 507 * Cleans a subdirectory under the test data directory on the test filesystem. 508 * @return True if we removed child 509 */ 510 public boolean cleanupDataTestDirOnTestFS(String subdirName) throws IOException { 511 Path cpath = getDataTestDirOnTestFS(subdirName); 512 return getTestFileSystem().delete(cpath, true); 513 } 514 515 // Workaround to avoid IllegalThreadStateException 516 // See HBASE-27148 for more details 517 private static final class FsDatasetAsyncDiskServiceFixer extends Thread { 518 519 private volatile boolean stopped = false; 520 521 private final MiniDFSCluster cluster; 522 523 FsDatasetAsyncDiskServiceFixer(MiniDFSCluster cluster) { 524 super("FsDatasetAsyncDiskServiceFixer"); 525 setDaemon(true); 526 this.cluster = cluster; 527 } 528 529 @Override 530 public void run() { 531 while (!stopped) { 532 try { 533 Thread.sleep(30000); 534 } catch (InterruptedException e) { 535 Thread.currentThread().interrupt(); 536 continue; 537 } 538 // we could add new datanodes during tests, so here we will check every 30 seconds, as the 539 // timeout of the thread pool executor is 60 seconds by default. 540 try { 541 for (DataNode dn : cluster.getDataNodes()) { 542 FsDatasetSpi<?> dataset = dn.getFSDataset(); 543 Field service = dataset.getClass().getDeclaredField("asyncDiskService"); 544 service.setAccessible(true); 545 Object asyncDiskService = service.get(dataset); 546 Field group = asyncDiskService.getClass().getDeclaredField("threadGroup"); 547 group.setAccessible(true); 548 ThreadGroup threadGroup = (ThreadGroup) group.get(asyncDiskService); 549 if (threadGroup.isDaemon()) { 550 threadGroup.setDaemon(false); 551 } 552 } 553 } catch (NoSuchFieldException e) { 554 LOG.debug("NoSuchFieldException: " + e.getMessage() 555 + "; It might because your Hadoop version > 3.2.3 or 3.3.4, " 556 + "See HBASE-27595 for details."); 557 } catch (Exception e) { 558 LOG.warn("failed to reset thread pool timeout for FsDatasetAsyncDiskService", e); 559 } 560 } 561 } 562 563 void shutdown() { 564 stopped = true; 565 interrupt(); 566 } 567 } 568 569 /** 570 * Start a minidfscluster. 571 * @param servers How many DNs to start. 572 * @see #shutdownMiniDFSCluster() 573 * @return The mini dfs cluster created. 574 */ 575 public MiniDFSCluster startMiniDFSCluster(int servers) throws Exception { 576 return startMiniDFSCluster(servers, null); 577 } 578 579 /** 580 * Start a minidfscluster. This is useful if you want to run datanode on distinct hosts for things 581 * like HDFS block location verification. If you start MiniDFSCluster without host names, all 582 * instances of the datanodes will have the same host name. 583 * @param hosts hostnames DNs to run on. 584 * @see #shutdownMiniDFSCluster() 585 * @return The mini dfs cluster created. 586 */ 587 public MiniDFSCluster startMiniDFSCluster(final String hosts[]) throws Exception { 588 if (hosts != null && hosts.length != 0) { 589 return startMiniDFSCluster(hosts.length, hosts); 590 } else { 591 return startMiniDFSCluster(1, null); 592 } 593 } 594 595 /** 596 * Start a minidfscluster. Can only create one. 597 * @param servers How many DNs to start. 598 * @param hosts hostnames DNs to run on. 599 * @see #shutdownMiniDFSCluster() 600 * @return The mini dfs cluster created. 601 */ 602 public MiniDFSCluster startMiniDFSCluster(int servers, final String hosts[]) throws Exception { 603 return startMiniDFSCluster(servers, null, hosts); 604 } 605 606 private void setFs() throws IOException { 607 if (this.dfsCluster == null) { 608 LOG.info("Skipping setting fs because dfsCluster is null"); 609 return; 610 } 611 FileSystem fs = this.dfsCluster.getFileSystem(); 612 CommonFSUtils.setFsDefault(this.conf, new Path(fs.getUri())); 613 614 // re-enable this check with dfs 615 conf.unset(CommonFSUtils.UNSAFE_STREAM_CAPABILITY_ENFORCE); 616 } 617 618 public MiniDFSCluster startMiniDFSCluster(int servers, final String racks[], String hosts[]) 619 throws Exception { 620 createDirsAndSetProperties(); 621 EditLogFileOutputStream.setShouldSkipFsyncForTesting(true); 622 623 this.dfsCluster = 624 new MiniDFSCluster(0, this.conf, servers, true, true, true, null, racks, hosts, null); 625 this.dfsClusterFixer = new FsDatasetAsyncDiskServiceFixer(dfsCluster); 626 this.dfsClusterFixer.start(); 627 // Set this just-started cluster as our filesystem. 628 setFs(); 629 630 // Wait for the cluster to be totally up 631 this.dfsCluster.waitClusterUp(); 632 633 // reset the test directory for test file system 634 dataTestDirOnTestFS = null; 635 String dataTestDir = getDataTestDir().toString(); 636 conf.set(HConstants.HBASE_DIR, dataTestDir); 637 LOG.debug("Setting {} to {}", HConstants.HBASE_DIR, dataTestDir); 638 639 return this.dfsCluster; 640 } 641 642 public MiniDFSCluster startMiniDFSClusterForTestWAL(int namenodePort) throws IOException { 643 createDirsAndSetProperties(); 644 dfsCluster = 645 new MiniDFSCluster(namenodePort, conf, 5, false, true, true, null, null, null, null); 646 this.dfsClusterFixer = new FsDatasetAsyncDiskServiceFixer(dfsCluster); 647 this.dfsClusterFixer.start(); 648 return dfsCluster; 649 } 650 651 /** 652 * This is used before starting HDFS and map-reduce mini-clusters Run something like the below to 653 * check for the likes of '/tmp' references -- i.e. references outside of the test data dir -- in 654 * the conf. 655 * 656 * <pre> 657 * Configuration conf = TEST_UTIL.getConfiguration(); 658 * for (Iterator<Map.Entry<String, String>> i = conf.iterator(); i.hasNext();) { 659 * Map.Entry<String, String> e = i.next(); 660 * assertFalse(e.getKey() + " " + e.getValue(), e.getValue().contains("/tmp")); 661 * } 662 * </pre> 663 */ 664 private void createDirsAndSetProperties() throws IOException { 665 setupClusterTestDir(); 666 conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, clusterTestDir.getCanonicalPath()); 667 createDirAndSetProperty("test.cache.data"); 668 createDirAndSetProperty("hadoop.tmp.dir"); 669 hadoopLogDir = createDirAndSetProperty("hadoop.log.dir"); 670 createDirAndSetProperty("mapreduce.cluster.local.dir"); 671 createDirAndSetProperty("mapreduce.cluster.temp.dir"); 672 enableShortCircuit(); 673 674 Path root = getDataTestDirOnTestFS("hadoop"); 675 conf.set(MapreduceTestingShim.getMROutputDirProp(), 676 new Path(root, "mapred-output-dir").toString()); 677 conf.set("mapreduce.jobtracker.system.dir", new Path(root, "mapred-system-dir").toString()); 678 conf.set("mapreduce.jobtracker.staging.root.dir", 679 new Path(root, "mapreduce-jobtracker-staging-root-dir").toString()); 680 conf.set("mapreduce.job.working.dir", new Path(root, "mapred-working-dir").toString()); 681 conf.set("yarn.app.mapreduce.am.staging-dir", 682 new Path(root, "mapreduce-am-staging-root-dir").toString()); 683 684 // Frustrate yarn's and hdfs's attempts at writing /tmp. 685 // Below is fragile. Make it so we just interpolate any 'tmp' reference. 686 createDirAndSetProperty("yarn.node-labels.fs-store.root-dir"); 687 createDirAndSetProperty("yarn.node-attribute.fs-store.root-dir"); 688 createDirAndSetProperty("yarn.nodemanager.log-dirs"); 689 createDirAndSetProperty("yarn.nodemanager.remote-app-log-dir"); 690 createDirAndSetProperty("yarn.timeline-service.entity-group-fs-store.active-dir"); 691 createDirAndSetProperty("yarn.timeline-service.entity-group-fs-store.done-dir"); 692 createDirAndSetProperty("yarn.nodemanager.remote-app-log-dir"); 693 createDirAndSetProperty("dfs.journalnode.edits.dir"); 694 createDirAndSetProperty("dfs.datanode.shared.file.descriptor.paths"); 695 createDirAndSetProperty("nfs.dump.dir"); 696 createDirAndSetProperty("java.io.tmpdir"); 697 createDirAndSetProperty("dfs.journalnode.edits.dir"); 698 createDirAndSetProperty("dfs.provided.aliasmap.inmemory.leveldb.dir"); 699 createDirAndSetProperty("fs.s3a.committer.staging.tmp.path"); 700 701 // disable metrics logger since it depend on commons-logging internal classes and we do not want 702 // commons-logging on our classpath 703 conf.setInt(DFSConfigKeys.DFS_NAMENODE_METRICS_LOGGER_PERIOD_SECONDS_KEY, 0); 704 conf.setInt(DFSConfigKeys.DFS_DATANODE_METRICS_LOGGER_PERIOD_SECONDS_KEY, 0); 705 } 706 707 /** 708 * Check whether the tests should assume NEW_VERSION_BEHAVIOR when creating new column families. 709 * Default to false. 710 */ 711 public boolean isNewVersionBehaviorEnabled() { 712 final String propName = "hbase.tests.new.version.behavior"; 713 String v = System.getProperty(propName); 714 if (v != null) { 715 return Boolean.parseBoolean(v); 716 } 717 return false; 718 } 719 720 /** 721 * Get the HBase setting for dfs.client.read.shortcircuit from the conf or a system property. This 722 * allows to specify this parameter on the command line. If not set, default is true. 723 */ 724 public boolean isReadShortCircuitOn() { 725 final String propName = "hbase.tests.use.shortcircuit.reads"; 726 String readOnProp = System.getProperty(propName); 727 if (readOnProp != null) { 728 return Boolean.parseBoolean(readOnProp); 729 } else { 730 return conf.getBoolean(propName, false); 731 } 732 } 733 734 /** 735 * Enable the short circuit read, unless configured differently. Set both HBase and HDFS settings, 736 * including skipping the hdfs checksum checks. 737 */ 738 private void enableShortCircuit() { 739 if (isReadShortCircuitOn()) { 740 String curUser = System.getProperty("user.name"); 741 LOG.info("read short circuit is ON for user " + curUser); 742 // read short circuit, for hdfs 743 conf.set("dfs.block.local-path-access.user", curUser); 744 // read short circuit, for hbase 745 conf.setBoolean("dfs.client.read.shortcircuit", true); 746 // Skip checking checksum, for the hdfs client and the datanode 747 conf.setBoolean("dfs.client.read.shortcircuit.skip.checksum", true); 748 } else { 749 LOG.info("read short circuit is OFF"); 750 } 751 } 752 753 private String createDirAndSetProperty(final String property) { 754 return createDirAndSetProperty(property, property); 755 } 756 757 private String createDirAndSetProperty(final String relPath, String property) { 758 String path = getDataTestDir(relPath).toString(); 759 System.setProperty(property, path); 760 conf.set(property, path); 761 new File(path).mkdirs(); 762 LOG.info("Setting " + property + " to " + path + " in system properties and HBase conf"); 763 return path; 764 } 765 766 /** 767 * Shuts down instance created by call to {@link #startMiniDFSCluster(int)} or does nothing. 768 */ 769 public void shutdownMiniDFSCluster() throws IOException { 770 if (this.dfsCluster != null) { 771 // The below throws an exception per dn, AsynchronousCloseException. 772 this.dfsCluster.shutdown(); 773 dfsCluster = null; 774 // It is possible that the dfs cluster is set through setDFSCluster method, where we will not 775 // have a fixer 776 if (dfsClusterFixer != null) { 777 this.dfsClusterFixer.shutdown(); 778 dfsClusterFixer = null; 779 } 780 dataTestDirOnTestFS = null; 781 CommonFSUtils.setFsDefault(this.conf, new Path("file:///")); 782 } 783 } 784 785 /** 786 * Start up a minicluster of hbase, dfs, and zookeeper where WAL's walDir is created separately. 787 * All other options will use default values, defined in {@link StartMiniClusterOption.Builder}. 788 * @param createWALDir Whether to create a new WAL directory. 789 * @return The mini HBase cluster created. 790 * @see #shutdownMiniCluster() 791 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 792 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 793 * @see #startMiniCluster(StartMiniClusterOption) 794 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 795 */ 796 @Deprecated 797 public MiniHBaseCluster startMiniCluster(boolean createWALDir) throws Exception { 798 StartMiniClusterOption option = 799 StartMiniClusterOption.builder().createWALDir(createWALDir).build(); 800 return startMiniCluster(option); 801 } 802 803 /** 804 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 805 * defined in {@link StartMiniClusterOption.Builder}. 806 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 807 * @param createRootDir Whether to create a new root or data directory path. 808 * @return The mini HBase cluster created. 809 * @see #shutdownMiniCluster() 810 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 811 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 812 * @see #startMiniCluster(StartMiniClusterOption) 813 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 814 */ 815 @Deprecated 816 public MiniHBaseCluster startMiniCluster(int numSlaves, boolean createRootDir) throws Exception { 817 StartMiniClusterOption option = StartMiniClusterOption.builder().numRegionServers(numSlaves) 818 .numDataNodes(numSlaves).createRootDir(createRootDir).build(); 819 return startMiniCluster(option); 820 } 821 822 /** 823 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 824 * defined in {@link StartMiniClusterOption.Builder}. 825 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 826 * @param createRootDir Whether to create a new root or data directory path. 827 * @param createWALDir Whether to create a new WAL directory. 828 * @return The mini HBase cluster created. 829 * @see #shutdownMiniCluster() 830 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 831 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 832 * @see #startMiniCluster(StartMiniClusterOption) 833 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 834 */ 835 @Deprecated 836 public MiniHBaseCluster startMiniCluster(int numSlaves, boolean createRootDir, 837 boolean createWALDir) throws Exception { 838 StartMiniClusterOption option = StartMiniClusterOption.builder().numRegionServers(numSlaves) 839 .numDataNodes(numSlaves).createRootDir(createRootDir).createWALDir(createWALDir).build(); 840 return startMiniCluster(option); 841 } 842 843 /** 844 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 845 * defined in {@link StartMiniClusterOption.Builder}. 846 * @param numMasters Master node number. 847 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 848 * @param createRootDir Whether to create a new root or data directory path. 849 * @return The mini HBase cluster created. 850 * @see #shutdownMiniCluster() 851 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 852 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 853 * @see #startMiniCluster(StartMiniClusterOption) 854 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 855 */ 856 @Deprecated 857 public MiniHBaseCluster startMiniCluster(int numMasters, int numSlaves, boolean createRootDir) 858 throws Exception { 859 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 860 .numRegionServers(numSlaves).createRootDir(createRootDir).numDataNodes(numSlaves).build(); 861 return startMiniCluster(option); 862 } 863 864 /** 865 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 866 * defined in {@link StartMiniClusterOption.Builder}. 867 * @param numMasters Master node number. 868 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 869 * @return The mini HBase cluster created. 870 * @see #shutdownMiniCluster() 871 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 872 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 873 * @see #startMiniCluster(StartMiniClusterOption) 874 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 875 */ 876 @Deprecated 877 public MiniHBaseCluster startMiniCluster(int numMasters, int numSlaves) throws Exception { 878 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 879 .numRegionServers(numSlaves).numDataNodes(numSlaves).build(); 880 return startMiniCluster(option); 881 } 882 883 /** 884 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 885 * defined in {@link StartMiniClusterOption.Builder}. 886 * @param numMasters Master node number. 887 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 888 * @param dataNodeHosts The hostnames of DataNodes to run on. If not null, its size will overwrite 889 * HDFS data node number. 890 * @param createRootDir Whether to create a new root or data directory path. 891 * @return The mini HBase cluster created. 892 * @see #shutdownMiniCluster() 893 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 894 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 895 * @see #startMiniCluster(StartMiniClusterOption) 896 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 897 */ 898 @Deprecated 899 public MiniHBaseCluster startMiniCluster(int numMasters, int numSlaves, String[] dataNodeHosts, 900 boolean createRootDir) throws Exception { 901 StartMiniClusterOption option = 902 StartMiniClusterOption.builder().numMasters(numMasters).numRegionServers(numSlaves) 903 .createRootDir(createRootDir).numDataNodes(numSlaves).dataNodeHosts(dataNodeHosts).build(); 904 return startMiniCluster(option); 905 } 906 907 /** 908 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 909 * defined in {@link StartMiniClusterOption.Builder}. 910 * @param numMasters Master node number. 911 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 912 * @param dataNodeHosts The hostnames of DataNodes to run on. If not null, its size will overwrite 913 * HDFS data node number. 914 * @return The mini HBase cluster created. 915 * @see #shutdownMiniCluster() 916 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 917 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 918 * @see #startMiniCluster(StartMiniClusterOption) 919 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 920 */ 921 @Deprecated 922 public MiniHBaseCluster startMiniCluster(int numMasters, int numSlaves, String[] dataNodeHosts) 923 throws Exception { 924 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 925 .numRegionServers(numSlaves).numDataNodes(numSlaves).dataNodeHosts(dataNodeHosts).build(); 926 return startMiniCluster(option); 927 } 928 929 /** 930 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 931 * defined in {@link StartMiniClusterOption.Builder}. 932 * @param numMasters Master node number. 933 * @param numRegionServers Number of region servers. 934 * @param numDataNodes Number of datanodes. 935 * @return The mini HBase cluster created. 936 * @see #shutdownMiniCluster() 937 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 938 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 939 * @see #startMiniCluster(StartMiniClusterOption) 940 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 941 */ 942 @Deprecated 943 public MiniHBaseCluster startMiniCluster(int numMasters, int numRegionServers, int numDataNodes) 944 throws Exception { 945 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 946 .numRegionServers(numRegionServers).numDataNodes(numDataNodes).build(); 947 return startMiniCluster(option); 948 } 949 950 /** 951 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 952 * defined in {@link StartMiniClusterOption.Builder}. 953 * @param numMasters Master node number. 954 * @param numSlaves Slave node number, for both HBase region server and HDFS data node. 955 * @param dataNodeHosts The hostnames of DataNodes to run on. If not null, its size will overwrite 956 * HDFS data node number. 957 * @param masterClass The class to use as HMaster, or null for default. 958 * @param rsClass The class to use as HRegionServer, or null for default. 959 * @return The mini HBase cluster created. 960 * @see #shutdownMiniCluster() 961 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 962 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 963 * @see #startMiniCluster(StartMiniClusterOption) 964 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 965 */ 966 @Deprecated 967 public MiniHBaseCluster startMiniCluster(int numMasters, int numSlaves, String[] dataNodeHosts, 968 Class<? extends HMaster> masterClass, 969 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> rsClass) throws Exception { 970 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 971 .masterClass(masterClass).numRegionServers(numSlaves).rsClass(rsClass).numDataNodes(numSlaves) 972 .dataNodeHosts(dataNodeHosts).build(); 973 return startMiniCluster(option); 974 } 975 976 /** 977 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 978 * defined in {@link StartMiniClusterOption.Builder}. 979 * @param numMasters Master node number. 980 * @param numRegionServers Number of region servers. 981 * @param numDataNodes Number of datanodes. 982 * @param dataNodeHosts The hostnames of DataNodes to run on. If not null, its size will 983 * overwrite HDFS data node number. 984 * @param masterClass The class to use as HMaster, or null for default. 985 * @param rsClass The class to use as HRegionServer, or null for default. 986 * @return The mini HBase cluster created. 987 * @see #shutdownMiniCluster() 988 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 989 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 990 * @see #startMiniCluster(StartMiniClusterOption) 991 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 992 */ 993 @Deprecated 994 public MiniHBaseCluster startMiniCluster(int numMasters, int numRegionServers, int numDataNodes, 995 String[] dataNodeHosts, Class<? extends HMaster> masterClass, 996 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> rsClass) throws Exception { 997 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 998 .masterClass(masterClass).numRegionServers(numRegionServers).rsClass(rsClass) 999 .numDataNodes(numDataNodes).dataNodeHosts(dataNodeHosts).build(); 1000 return startMiniCluster(option); 1001 } 1002 1003 /** 1004 * Start up a minicluster of hbase, dfs, and zookeeper. All other options will use default values, 1005 * defined in {@link StartMiniClusterOption.Builder}. 1006 * @param numMasters Master node number. 1007 * @param numRegionServers Number of region servers. 1008 * @param numDataNodes Number of datanodes. 1009 * @param dataNodeHosts The hostnames of DataNodes to run on. If not null, its size will 1010 * overwrite HDFS data node number. 1011 * @param masterClass The class to use as HMaster, or null for default. 1012 * @param rsClass The class to use as HRegionServer, or null for default. 1013 * @param createRootDir Whether to create a new root or data directory path. 1014 * @param createWALDir Whether to create a new WAL directory. 1015 * @return The mini HBase cluster created. 1016 * @see #shutdownMiniCluster() 1017 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 1018 * {@link #startMiniCluster(StartMiniClusterOption)} instead. 1019 * @see #startMiniCluster(StartMiniClusterOption) 1020 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 1021 */ 1022 @Deprecated 1023 public MiniHBaseCluster startMiniCluster(int numMasters, int numRegionServers, int numDataNodes, 1024 String[] dataNodeHosts, Class<? extends HMaster> masterClass, 1025 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> rsClass, boolean createRootDir, 1026 boolean createWALDir) throws Exception { 1027 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 1028 .masterClass(masterClass).numRegionServers(numRegionServers).rsClass(rsClass) 1029 .numDataNodes(numDataNodes).dataNodeHosts(dataNodeHosts).createRootDir(createRootDir) 1030 .createWALDir(createWALDir).build(); 1031 return startMiniCluster(option); 1032 } 1033 1034 /** 1035 * Start up a minicluster of hbase, dfs and zookeeper clusters with given slave node number. All 1036 * other options will use default values, defined in {@link StartMiniClusterOption.Builder}. 1037 * @param numSlaves slave node number, for both HBase region server and HDFS data node. 1038 * @see #startMiniCluster(StartMiniClusterOption option) 1039 * @see #shutdownMiniDFSCluster() 1040 */ 1041 public MiniHBaseCluster startMiniCluster(int numSlaves) throws Exception { 1042 StartMiniClusterOption option = 1043 StartMiniClusterOption.builder().numRegionServers(numSlaves).numDataNodes(numSlaves).build(); 1044 return startMiniCluster(option); 1045 } 1046 1047 /** 1048 * Start up a minicluster of hbase, dfs and zookeeper all using default options. Option default 1049 * value can be found in {@link StartMiniClusterOption.Builder}. 1050 * @see #startMiniCluster(StartMiniClusterOption option) 1051 * @see #shutdownMiniDFSCluster() 1052 */ 1053 public MiniHBaseCluster startMiniCluster() throws Exception { 1054 return startMiniCluster(StartMiniClusterOption.builder().build()); 1055 } 1056 1057 /** 1058 * Start up a mini cluster of hbase, optionally dfs and zookeeper if needed. It modifies 1059 * Configuration. It homes the cluster data directory under a random subdirectory in a directory 1060 * under System property test.build.data, to be cleaned up on exit. 1061 * @see #shutdownMiniDFSCluster() 1062 */ 1063 public MiniHBaseCluster startMiniCluster(StartMiniClusterOption option) throws Exception { 1064 LOG.info("Starting up minicluster with option: {}", option); 1065 1066 // If we already put up a cluster, fail. 1067 if (miniClusterRunning) { 1068 throw new IllegalStateException("A mini-cluster is already running"); 1069 } 1070 miniClusterRunning = true; 1071 1072 setupClusterTestDir(); 1073 1074 // Bring up mini dfs cluster. This spews a bunch of warnings about missing 1075 // scheme. Complaints are 'Scheme is undefined for build/test/data/dfs/name1'. 1076 if (dfsCluster == null) { 1077 LOG.info("STARTING DFS"); 1078 dfsCluster = startMiniDFSCluster(option.getNumDataNodes(), option.getDataNodeHosts()); 1079 } else { 1080 LOG.info("NOT STARTING DFS"); 1081 } 1082 1083 // Start up a zk cluster. 1084 if (getZkCluster() == null) { 1085 startMiniZKCluster(option.getNumZkServers()); 1086 } 1087 1088 // Start the MiniHBaseCluster 1089 return startMiniHBaseCluster(option); 1090 } 1091 1092 /** 1093 * Starts up mini hbase cluster. Usually you won't want this. You'll usually want 1094 * {@link #startMiniCluster()}. This is useful when doing stepped startup of clusters. 1095 * @return Reference to the hbase mini hbase cluster. 1096 * @see #startMiniCluster(StartMiniClusterOption) 1097 * @see #shutdownMiniHBaseCluster() 1098 */ 1099 public MiniHBaseCluster startMiniHBaseCluster(StartMiniClusterOption option) 1100 throws IOException, InterruptedException { 1101 // Now do the mini hbase cluster. Set the hbase.rootdir in config. 1102 createRootDir(option.isCreateRootDir()); 1103 if (option.isCreateWALDir()) { 1104 createWALRootDir(); 1105 } 1106 // Set the hbase.fs.tmp.dir config to make sure that we have some default value. This is 1107 // for tests that do not read hbase-defaults.xml 1108 setHBaseFsTmpDir(); 1109 1110 // These settings will make the server waits until this exact number of 1111 // regions servers are connected. 1112 if (conf.getInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, -1) == -1) { 1113 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, option.getNumRegionServers()); 1114 } 1115 if (conf.getInt(ServerManager.WAIT_ON_REGIONSERVERS_MAXTOSTART, -1) == -1) { 1116 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MAXTOSTART, option.getNumRegionServers()); 1117 } 1118 1119 Configuration c = new Configuration(this.conf); 1120 this.hbaseCluster = new MiniHBaseCluster(c, option.getNumMasters(), 1121 option.getNumAlwaysStandByMasters(), option.getNumRegionServers(), option.getRsPorts(), 1122 option.getMasterClass(), option.getRsClass()); 1123 // Populate the master address configuration from mini cluster configuration. 1124 conf.set(HConstants.MASTER_ADDRS_KEY, MasterRegistry.getMasterAddr(c)); 1125 // Don't leave here till we've done a successful scan of the hbase:meta 1126 try (Table t = getConnection().getTable(TableName.META_TABLE_NAME); 1127 ResultScanner s = t.getScanner(new Scan())) { 1128 for (;;) { 1129 if (s.next() == null) { 1130 break; 1131 } 1132 } 1133 } 1134 1135 getAdmin(); // create immediately the hbaseAdmin 1136 LOG.info("Minicluster is up; activeMaster={}", getHBaseCluster().getMaster()); 1137 1138 return (MiniHBaseCluster) hbaseCluster; 1139 } 1140 1141 /** 1142 * Starts up mini hbase cluster using default options. Default options can be found in 1143 * {@link StartMiniClusterOption.Builder}. 1144 * @see #startMiniHBaseCluster(StartMiniClusterOption) 1145 * @see #shutdownMiniHBaseCluster() 1146 */ 1147 public MiniHBaseCluster startMiniHBaseCluster() throws IOException, InterruptedException { 1148 return startMiniHBaseCluster(StartMiniClusterOption.builder().build()); 1149 } 1150 1151 /** 1152 * Starts up mini hbase cluster. Usually you won't want this. You'll usually want 1153 * {@link #startMiniCluster()}. All other options will use default values, defined in 1154 * {@link StartMiniClusterOption.Builder}. 1155 * @param numMasters Master node number. 1156 * @param numRegionServers Number of region servers. 1157 * @return The mini HBase cluster created. 1158 * @see #shutdownMiniHBaseCluster() 1159 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 1160 * {@link #startMiniHBaseCluster(StartMiniClusterOption)} instead. 1161 * @see #startMiniHBaseCluster(StartMiniClusterOption) 1162 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 1163 */ 1164 @Deprecated 1165 public MiniHBaseCluster startMiniHBaseCluster(int numMasters, int numRegionServers) 1166 throws IOException, InterruptedException { 1167 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 1168 .numRegionServers(numRegionServers).build(); 1169 return startMiniHBaseCluster(option); 1170 } 1171 1172 /** 1173 * Starts up mini hbase cluster. Usually you won't want this. You'll usually want 1174 * {@link #startMiniCluster()}. All other options will use default values, defined in 1175 * {@link StartMiniClusterOption.Builder}. 1176 * @param numMasters Master node number. 1177 * @param numRegionServers Number of region servers. 1178 * @param rsPorts Ports that RegionServer should use. 1179 * @return The mini HBase cluster created. 1180 * @see #shutdownMiniHBaseCluster() 1181 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 1182 * {@link #startMiniHBaseCluster(StartMiniClusterOption)} instead. 1183 * @see #startMiniHBaseCluster(StartMiniClusterOption) 1184 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 1185 */ 1186 @Deprecated 1187 public MiniHBaseCluster startMiniHBaseCluster(int numMasters, int numRegionServers, 1188 List<Integer> rsPorts) throws IOException, InterruptedException { 1189 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 1190 .numRegionServers(numRegionServers).rsPorts(rsPorts).build(); 1191 return startMiniHBaseCluster(option); 1192 } 1193 1194 /** 1195 * Starts up mini hbase cluster. Usually you won't want this. You'll usually want 1196 * {@link #startMiniCluster()}. All other options will use default values, defined in 1197 * {@link StartMiniClusterOption.Builder}. 1198 * @param numMasters Master node number. 1199 * @param numRegionServers Number of region servers. 1200 * @param rsPorts Ports that RegionServer should use. 1201 * @param masterClass The class to use as HMaster, or null for default. 1202 * @param rsClass The class to use as HRegionServer, or null for default. 1203 * @param createRootDir Whether to create a new root or data directory path. 1204 * @param createWALDir Whether to create a new WAL directory. 1205 * @return The mini HBase cluster created. 1206 * @see #shutdownMiniHBaseCluster() 1207 * @deprecated since 2.2.0 and will be removed in 4.0.0. Use 1208 * {@link #startMiniHBaseCluster(StartMiniClusterOption)} instead. 1209 * @see #startMiniHBaseCluster(StartMiniClusterOption) 1210 * @see <a href="https://issues.apache.org/jira/browse/HBASE-21071">HBASE-21071</a> 1211 */ 1212 @Deprecated 1213 public MiniHBaseCluster startMiniHBaseCluster(int numMasters, int numRegionServers, 1214 List<Integer> rsPorts, Class<? extends HMaster> masterClass, 1215 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> rsClass, boolean createRootDir, 1216 boolean createWALDir) throws IOException, InterruptedException { 1217 StartMiniClusterOption option = StartMiniClusterOption.builder().numMasters(numMasters) 1218 .masterClass(masterClass).numRegionServers(numRegionServers).rsClass(rsClass).rsPorts(rsPorts) 1219 .createRootDir(createRootDir).createWALDir(createWALDir).build(); 1220 return startMiniHBaseCluster(option); 1221 } 1222 1223 /** 1224 * Starts the hbase cluster up again after shutting it down previously in a test. Use this if you 1225 * want to keep dfs/zk up and just stop/start hbase. 1226 * @param servers number of region servers 1227 */ 1228 public void restartHBaseCluster(int servers) throws IOException, InterruptedException { 1229 this.restartHBaseCluster(servers, null); 1230 } 1231 1232 public void restartHBaseCluster(int servers, List<Integer> ports) 1233 throws IOException, InterruptedException { 1234 StartMiniClusterOption option = 1235 StartMiniClusterOption.builder().numRegionServers(servers).rsPorts(ports).build(); 1236 restartHBaseCluster(option); 1237 invalidateConnection(); 1238 } 1239 1240 public void restartHBaseCluster(StartMiniClusterOption option) 1241 throws IOException, InterruptedException { 1242 closeConnection(); 1243 this.hbaseCluster = new MiniHBaseCluster(this.conf, option.getNumMasters(), 1244 option.getNumAlwaysStandByMasters(), option.getNumRegionServers(), option.getRsPorts(), 1245 option.getMasterClass(), option.getRsClass()); 1246 // Don't leave here till we've done a successful scan of the hbase:meta 1247 Connection conn = ConnectionFactory.createConnection(this.conf); 1248 Table t = conn.getTable(TableName.META_TABLE_NAME); 1249 ResultScanner s = t.getScanner(new Scan()); 1250 while (s.next() != null) { 1251 // do nothing 1252 } 1253 LOG.info("HBase has been restarted"); 1254 s.close(); 1255 t.close(); 1256 conn.close(); 1257 } 1258 1259 /** 1260 * @return Current mini hbase cluster. Only has something in it after a call to 1261 * {@link #startMiniCluster()}. 1262 * @see #startMiniCluster() 1263 */ 1264 public MiniHBaseCluster getMiniHBaseCluster() { 1265 if (this.hbaseCluster == null || this.hbaseCluster instanceof MiniHBaseCluster) { 1266 return (MiniHBaseCluster) this.hbaseCluster; 1267 } 1268 throw new RuntimeException( 1269 hbaseCluster + " not an instance of " + MiniHBaseCluster.class.getName()); 1270 } 1271 1272 /** 1273 * Stops mini hbase, zk, and hdfs clusters. 1274 * @see #startMiniCluster(int) 1275 */ 1276 public void shutdownMiniCluster() throws IOException { 1277 LOG.info("Shutting down minicluster"); 1278 shutdownMiniHBaseCluster(); 1279 shutdownMiniDFSCluster(); 1280 shutdownMiniZKCluster(); 1281 1282 cleanupTestDir(); 1283 miniClusterRunning = false; 1284 LOG.info("Minicluster is down"); 1285 } 1286 1287 /** 1288 * Shutdown HBase mini cluster.Does not shutdown zk or dfs if running. 1289 * @throws java.io.IOException in case command is unsuccessful 1290 */ 1291 public void shutdownMiniHBaseCluster() throws IOException { 1292 cleanup(); 1293 if (this.hbaseCluster != null) { 1294 this.hbaseCluster.shutdown(); 1295 // Wait till hbase is down before going on to shutdown zk. 1296 this.hbaseCluster.waitUntilShutDown(); 1297 this.hbaseCluster = null; 1298 } 1299 if (zooKeeperWatcher != null) { 1300 zooKeeperWatcher.close(); 1301 zooKeeperWatcher = null; 1302 } 1303 } 1304 1305 /** 1306 * Abruptly Shutdown HBase mini cluster. Does not shutdown zk or dfs if running. 1307 * @throws java.io.IOException throws in case command is unsuccessful 1308 */ 1309 public void killMiniHBaseCluster() throws IOException { 1310 cleanup(); 1311 if (this.hbaseCluster != null) { 1312 getMiniHBaseCluster().killAll(); 1313 this.hbaseCluster = null; 1314 } 1315 if (zooKeeperWatcher != null) { 1316 zooKeeperWatcher.close(); 1317 zooKeeperWatcher = null; 1318 } 1319 } 1320 1321 // close hbase admin, close current connection and reset MIN MAX configs for RS. 1322 private void cleanup() throws IOException { 1323 closeConnection(); 1324 // unset the configuration for MIN and MAX RS to start 1325 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, -1); 1326 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MAXTOSTART, -1); 1327 } 1328 1329 /** 1330 * Returns the path to the default root dir the minicluster uses. If <code>create</code> is true, 1331 * a new root directory path is fetched irrespective of whether it has been fetched before or not. 1332 * If false, previous path is used. Note: this does not cause the root dir to be created. 1333 * @return Fully qualified path for the default hbase root dir 1334 */ 1335 public Path getDefaultRootDirPath(boolean create) throws IOException { 1336 if (!create) { 1337 return getDataTestDirOnTestFS(); 1338 } else { 1339 return getNewDataTestDirOnTestFS(); 1340 } 1341 } 1342 1343 /** 1344 * Same as {{@link HBaseTestingUtility#getDefaultRootDirPath(boolean create)} except that 1345 * <code>create</code> flag is false. Note: this does not cause the root dir to be created. 1346 * @return Fully qualified path for the default hbase root dir 1347 */ 1348 public Path getDefaultRootDirPath() throws IOException { 1349 return getDefaultRootDirPath(false); 1350 } 1351 1352 /** 1353 * Creates an hbase rootdir in user home directory. Also creates hbase version file. Normally you 1354 * won't make use of this method. Root hbasedir is created for you as part of mini cluster 1355 * startup. You'd only use this method if you were doing manual operation. 1356 * @param create This flag decides whether to get a new root or data directory path or not, if it 1357 * has been fetched already. Note : Directory will be made irrespective of whether 1358 * path has been fetched or not. If directory already exists, it will be overwritten 1359 * @return Fully qualified path to hbase root dir 1360 */ 1361 public Path createRootDir(boolean create) throws IOException { 1362 FileSystem fs = FileSystem.get(this.conf); 1363 Path hbaseRootdir = getDefaultRootDirPath(create); 1364 CommonFSUtils.setRootDir(this.conf, hbaseRootdir); 1365 fs.mkdirs(hbaseRootdir); 1366 FSUtils.setVersion(fs, hbaseRootdir); 1367 return hbaseRootdir; 1368 } 1369 1370 /** 1371 * Same as {@link HBaseTestingUtility#createRootDir(boolean create)} except that 1372 * <code>create</code> flag is false. 1373 * @return Fully qualified path to hbase root dir 1374 */ 1375 public Path createRootDir() throws IOException { 1376 return createRootDir(false); 1377 } 1378 1379 /** 1380 * Creates a hbase walDir in the user's home directory. Normally you won't make use of this 1381 * method. Root hbaseWALDir is created for you as part of mini cluster startup. You'd only use 1382 * this method if you were doing manual operation. 1383 * @return Fully qualified path to hbase root dir 1384 */ 1385 public Path createWALRootDir() throws IOException { 1386 FileSystem fs = FileSystem.get(this.conf); 1387 Path walDir = getNewDataTestDirOnTestFS(); 1388 CommonFSUtils.setWALRootDir(this.conf, walDir); 1389 fs.mkdirs(walDir); 1390 return walDir; 1391 } 1392 1393 private void setHBaseFsTmpDir() throws IOException { 1394 String hbaseFsTmpDirInString = this.conf.get("hbase.fs.tmp.dir"); 1395 if (hbaseFsTmpDirInString == null) { 1396 this.conf.set("hbase.fs.tmp.dir", getDataTestDirOnTestFS("hbase-staging").toString()); 1397 LOG.info("Setting hbase.fs.tmp.dir to " + this.conf.get("hbase.fs.tmp.dir")); 1398 } else { 1399 LOG.info("The hbase.fs.tmp.dir is set to " + hbaseFsTmpDirInString); 1400 } 1401 } 1402 1403 /** 1404 * Flushes all caches in the mini hbase cluster 1405 */ 1406 public void flush() throws IOException { 1407 getMiniHBaseCluster().flushcache(); 1408 } 1409 1410 /** 1411 * Flushes all caches in the mini hbase cluster 1412 */ 1413 public void flush(TableName tableName) throws IOException { 1414 getMiniHBaseCluster().flushcache(tableName); 1415 } 1416 1417 /** 1418 * Compact all regions in the mini hbase cluster 1419 */ 1420 public void compact(boolean major) throws IOException { 1421 getMiniHBaseCluster().compact(major); 1422 } 1423 1424 /** 1425 * Compact all of a table's reagion in the mini hbase cluster 1426 */ 1427 public void compact(TableName tableName, boolean major) throws IOException { 1428 getMiniHBaseCluster().compact(tableName, major); 1429 } 1430 1431 /** 1432 * Create a table. 1433 * @return A Table instance for the created table. 1434 */ 1435 public Table createTable(TableName tableName, String family) throws IOException { 1436 return createTable(tableName, new String[] { family }); 1437 } 1438 1439 /** 1440 * Create a table. 1441 * @return A Table instance for the created table. 1442 */ 1443 public Table createTable(TableName tableName, String[] families) throws IOException { 1444 List<byte[]> fams = new ArrayList<>(families.length); 1445 for (String family : families) { 1446 fams.add(Bytes.toBytes(family)); 1447 } 1448 return createTable(tableName, fams.toArray(new byte[0][])); 1449 } 1450 1451 /** 1452 * Create a table. 1453 * @return A Table instance for the created table. 1454 */ 1455 public Table createTable(TableName tableName, byte[] family) throws IOException { 1456 return createTable(tableName, new byte[][] { family }); 1457 } 1458 1459 /** 1460 * Create a table with multiple regions. 1461 * @return A Table instance for the created table. 1462 */ 1463 public Table createMultiRegionTable(TableName tableName, byte[] family, int numRegions) 1464 throws IOException { 1465 if (numRegions < 3) throw new IOException("Must create at least 3 regions"); 1466 byte[] startKey = Bytes.toBytes("aaaaa"); 1467 byte[] endKey = Bytes.toBytes("zzzzz"); 1468 byte[][] splitKeys = Bytes.split(startKey, endKey, numRegions - 3); 1469 1470 return createTable(tableName, new byte[][] { family }, splitKeys); 1471 } 1472 1473 /** 1474 * Create a table. 1475 * @return A Table instance for the created table. 1476 */ 1477 public Table createTable(TableName tableName, byte[][] families) throws IOException { 1478 return createTable(tableName, families, (byte[][]) null); 1479 } 1480 1481 /** 1482 * Create a table with multiple regions. 1483 * @return A Table instance for the created table. 1484 */ 1485 public Table createMultiRegionTable(TableName tableName, byte[][] families) throws IOException { 1486 return createTable(tableName, families, KEYS_FOR_HBA_CREATE_TABLE); 1487 } 1488 1489 /** 1490 * Create a table with multiple regions. 1491 * @param replicaCount replica count. 1492 * @return A Table instance for the created table. 1493 */ 1494 public Table createMultiRegionTable(TableName tableName, int replicaCount, byte[][] families) 1495 throws IOException { 1496 return createTable(tableName, families, KEYS_FOR_HBA_CREATE_TABLE, replicaCount); 1497 } 1498 1499 /** 1500 * Create a table. 1501 * @return A Table instance for the created table. 1502 */ 1503 public Table createTable(TableName tableName, byte[][] families, byte[][] splitKeys) 1504 throws IOException { 1505 return createTable(tableName, families, splitKeys, 1, new Configuration(getConfiguration())); 1506 } 1507 1508 /** 1509 * Create a table. 1510 * @param tableName the table name 1511 * @param families the families 1512 * @param splitKeys the splitkeys 1513 * @param replicaCount the region replica count 1514 * @return A Table instance for the created table. 1515 * @throws IOException throws IOException 1516 */ 1517 public Table createTable(TableName tableName, byte[][] families, byte[][] splitKeys, 1518 int replicaCount) throws IOException { 1519 return createTable(tableName, families, splitKeys, replicaCount, 1520 new Configuration(getConfiguration())); 1521 } 1522 1523 public Table createTable(TableName tableName, byte[][] families, int numVersions, byte[] startKey, 1524 byte[] endKey, int numRegions) throws IOException { 1525 TableDescriptor desc = createTableDescriptor(tableName, families, numVersions); 1526 1527 getAdmin().createTable(desc, startKey, endKey, numRegions); 1528 // HBaseAdmin only waits for regions to appear in hbase:meta we 1529 // should wait until they are assigned 1530 waitUntilAllRegionsAssigned(tableName); 1531 return getConnection().getTable(tableName); 1532 } 1533 1534 /** 1535 * Create a table. 1536 * @param c Configuration to use 1537 * @return A Table instance for the created table. 1538 */ 1539 public Table createTable(TableDescriptor htd, byte[][] families, Configuration c) 1540 throws IOException { 1541 return createTable(htd, families, null, c); 1542 } 1543 1544 /** 1545 * Create a table. 1546 * @param htd table descriptor 1547 * @param families array of column families 1548 * @param splitKeys array of split keys 1549 * @param c Configuration to use 1550 * @return A Table instance for the created table. 1551 * @throws IOException if getAdmin or createTable fails 1552 */ 1553 public Table createTable(TableDescriptor htd, byte[][] families, byte[][] splitKeys, 1554 Configuration c) throws IOException { 1555 // Disable blooms (they are on by default as of 0.95) but we disable them here because 1556 // tests have hard coded counts of what to expect in block cache, etc., and blooms being 1557 // on is interfering. 1558 return createTable(htd, families, splitKeys, BloomType.NONE, HConstants.DEFAULT_BLOCKSIZE, c); 1559 } 1560 1561 /** 1562 * Create a table. 1563 * @param htd table descriptor 1564 * @param families array of column families 1565 * @param splitKeys array of split keys 1566 * @param type Bloom type 1567 * @param blockSize block size 1568 * @param c Configuration to use 1569 * @return A Table instance for the created table. 1570 * @throws IOException if getAdmin or createTable fails 1571 */ 1572 1573 public Table createTable(TableDescriptor htd, byte[][] families, byte[][] splitKeys, 1574 BloomType type, int blockSize, Configuration c) throws IOException { 1575 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(htd); 1576 for (byte[] family : families) { 1577 ColumnFamilyDescriptorBuilder cfdb = ColumnFamilyDescriptorBuilder.newBuilder(family) 1578 .setBloomFilterType(type).setBlocksize(blockSize); 1579 if (isNewVersionBehaviorEnabled()) { 1580 cfdb.setNewVersionBehavior(true); 1581 } 1582 builder.setColumnFamily(cfdb.build()); 1583 } 1584 TableDescriptor td = builder.build(); 1585 if (splitKeys != null) { 1586 getAdmin().createTable(td, splitKeys); 1587 } else { 1588 getAdmin().createTable(td); 1589 } 1590 // HBaseAdmin only waits for regions to appear in hbase:meta 1591 // we should wait until they are assigned 1592 waitUntilAllRegionsAssigned(td.getTableName()); 1593 return getConnection().getTable(td.getTableName()); 1594 } 1595 1596 /** 1597 * Create a table. 1598 * @param htd table descriptor 1599 * @param splitRows array of split keys 1600 * @return A Table instance for the created table. 1601 */ 1602 public Table createTable(TableDescriptor htd, byte[][] splitRows) throws IOException { 1603 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(htd); 1604 if (isNewVersionBehaviorEnabled()) { 1605 for (ColumnFamilyDescriptor family : htd.getColumnFamilies()) { 1606 builder.setColumnFamily( 1607 ColumnFamilyDescriptorBuilder.newBuilder(family).setNewVersionBehavior(true).build()); 1608 } 1609 } 1610 if (splitRows != null) { 1611 getAdmin().createTable(builder.build(), splitRows); 1612 } else { 1613 getAdmin().createTable(builder.build()); 1614 } 1615 // HBaseAdmin only waits for regions to appear in hbase:meta 1616 // we should wait until they are assigned 1617 waitUntilAllRegionsAssigned(htd.getTableName()); 1618 return getConnection().getTable(htd.getTableName()); 1619 } 1620 1621 /** 1622 * Create a table. 1623 * @param tableName the table name 1624 * @param families the families 1625 * @param splitKeys the split keys 1626 * @param replicaCount the replica count 1627 * @param c Configuration to use 1628 * @return A Table instance for the created table. 1629 */ 1630 public Table createTable(TableName tableName, byte[][] families, byte[][] splitKeys, 1631 int replicaCount, final Configuration c) throws IOException { 1632 TableDescriptor htd = 1633 TableDescriptorBuilder.newBuilder(tableName).setRegionReplication(replicaCount).build(); 1634 return createTable(htd, families, splitKeys, c); 1635 } 1636 1637 /** 1638 * Create a table. 1639 * @return A Table instance for the created table. 1640 */ 1641 public Table createTable(TableName tableName, byte[] family, int numVersions) throws IOException { 1642 return createTable(tableName, new byte[][] { family }, numVersions); 1643 } 1644 1645 /** 1646 * Create a table. 1647 * @return A Table instance for the created table. 1648 */ 1649 public Table createTable(TableName tableName, byte[][] families, int numVersions) 1650 throws IOException { 1651 return createTable(tableName, families, numVersions, (byte[][]) null); 1652 } 1653 1654 /** 1655 * Create a table. 1656 * @return A Table instance for the created table. 1657 */ 1658 public Table createTable(TableName tableName, byte[][] families, int numVersions, 1659 byte[][] splitKeys) throws IOException { 1660 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1661 for (byte[] family : families) { 1662 ColumnFamilyDescriptorBuilder cfBuilder = 1663 ColumnFamilyDescriptorBuilder.newBuilder(family).setMaxVersions(numVersions); 1664 if (isNewVersionBehaviorEnabled()) { 1665 cfBuilder.setNewVersionBehavior(true); 1666 } 1667 builder.setColumnFamily(cfBuilder.build()); 1668 } 1669 if (splitKeys != null) { 1670 getAdmin().createTable(builder.build(), splitKeys); 1671 } else { 1672 getAdmin().createTable(builder.build()); 1673 } 1674 // HBaseAdmin only waits for regions to appear in hbase:meta we should wait until they are 1675 // assigned 1676 waitUntilAllRegionsAssigned(tableName); 1677 return getConnection().getTable(tableName); 1678 } 1679 1680 /** 1681 * Create a table with multiple regions. 1682 * @return A Table instance for the created table. 1683 */ 1684 public Table createMultiRegionTable(TableName tableName, byte[][] families, int numVersions) 1685 throws IOException { 1686 return createTable(tableName, families, numVersions, KEYS_FOR_HBA_CREATE_TABLE); 1687 } 1688 1689 /** 1690 * Create a table. 1691 * @return A Table instance for the created table. 1692 */ 1693 public Table createTable(TableName tableName, byte[][] families, int numVersions, int blockSize) 1694 throws IOException { 1695 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1696 for (byte[] family : families) { 1697 ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(family) 1698 .setMaxVersions(numVersions).setBlocksize(blockSize); 1699 if (isNewVersionBehaviorEnabled()) { 1700 cfBuilder.setNewVersionBehavior(true); 1701 } 1702 builder.setColumnFamily(cfBuilder.build()); 1703 } 1704 getAdmin().createTable(builder.build()); 1705 // HBaseAdmin only waits for regions to appear in hbase:meta we should wait until they are 1706 // assigned 1707 waitUntilAllRegionsAssigned(tableName); 1708 return getConnection().getTable(tableName); 1709 } 1710 1711 public Table createTable(TableName tableName, byte[][] families, int numVersions, int blockSize, 1712 String cpName) throws IOException { 1713 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1714 for (byte[] family : families) { 1715 ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(family) 1716 .setMaxVersions(numVersions).setBlocksize(blockSize); 1717 if (isNewVersionBehaviorEnabled()) { 1718 cfBuilder.setNewVersionBehavior(true); 1719 } 1720 builder.setColumnFamily(cfBuilder.build()); 1721 } 1722 if (cpName != null) { 1723 builder.setCoprocessor(cpName); 1724 } 1725 getAdmin().createTable(builder.build()); 1726 // HBaseAdmin only waits for regions to appear in hbase:meta we should wait until they are 1727 // assigned 1728 waitUntilAllRegionsAssigned(tableName); 1729 return getConnection().getTable(tableName); 1730 } 1731 1732 /** 1733 * Create a table. 1734 * @return A Table instance for the created table. 1735 */ 1736 public Table createTable(TableName tableName, byte[][] families, int[] numVersions) 1737 throws IOException { 1738 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1739 int i = 0; 1740 for (byte[] family : families) { 1741 ColumnFamilyDescriptorBuilder cfBuilder = 1742 ColumnFamilyDescriptorBuilder.newBuilder(family).setMaxVersions(numVersions[i]); 1743 if (isNewVersionBehaviorEnabled()) { 1744 cfBuilder.setNewVersionBehavior(true); 1745 } 1746 builder.setColumnFamily(cfBuilder.build()); 1747 i++; 1748 } 1749 getAdmin().createTable(builder.build()); 1750 // HBaseAdmin only waits for regions to appear in hbase:meta we should wait until they are 1751 // assigned 1752 waitUntilAllRegionsAssigned(tableName); 1753 return getConnection().getTable(tableName); 1754 } 1755 1756 /** 1757 * Create a table. 1758 * @return A Table instance for the created table. 1759 */ 1760 public Table createTable(TableName tableName, byte[] family, byte[][] splitRows) 1761 throws IOException { 1762 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1763 ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(family); 1764 if (isNewVersionBehaviorEnabled()) { 1765 cfBuilder.setNewVersionBehavior(true); 1766 } 1767 builder.setColumnFamily(cfBuilder.build()); 1768 getAdmin().createTable(builder.build(), splitRows); 1769 // HBaseAdmin only waits for regions to appear in hbase:meta we should wait until they are 1770 // assigned 1771 waitUntilAllRegionsAssigned(tableName); 1772 return getConnection().getTable(tableName); 1773 } 1774 1775 /** 1776 * Create a table with multiple regions. 1777 * @return A Table instance for the created table. 1778 */ 1779 public Table createMultiRegionTable(TableName tableName, byte[] family) throws IOException { 1780 return createTable(tableName, family, KEYS_FOR_HBA_CREATE_TABLE); 1781 } 1782 1783 /** 1784 * Modify a table, synchronous. 1785 * @deprecated since 3.0.0 and will be removed in 4.0.0. Just use 1786 * {@link Admin#modifyTable(TableDescriptor)} directly as it is synchronous now. 1787 * @see Admin#modifyTable(TableDescriptor) 1788 * @see <a href="https://issues.apache.org/jira/browse/HBASE-22002">HBASE-22002</a> 1789 */ 1790 @Deprecated 1791 public static void modifyTableSync(Admin admin, TableDescriptor desc) 1792 throws IOException, InterruptedException { 1793 admin.modifyTable(desc); 1794 } 1795 1796 /** 1797 * Set the number of Region replicas. 1798 */ 1799 public static void setReplicas(Admin admin, TableName table, int replicaCount) 1800 throws IOException, InterruptedException { 1801 TableDescriptor desc = TableDescriptorBuilder.newBuilder(admin.getDescriptor(table)) 1802 .setRegionReplication(replicaCount).build(); 1803 admin.modifyTable(desc); 1804 } 1805 1806 /** 1807 * Drop an existing table 1808 * @param tableName existing table 1809 */ 1810 public void deleteTable(TableName tableName) throws IOException { 1811 try { 1812 getAdmin().disableTable(tableName); 1813 } catch (TableNotEnabledException e) { 1814 LOG.debug("Table: " + tableName + " already disabled, so just deleting it."); 1815 } 1816 getAdmin().deleteTable(tableName); 1817 } 1818 1819 /** 1820 * Drop an existing table 1821 * @param tableName existing table 1822 */ 1823 public void deleteTableIfAny(TableName tableName) throws IOException { 1824 try { 1825 deleteTable(tableName); 1826 } catch (TableNotFoundException e) { 1827 // ignore 1828 } 1829 } 1830 1831 // ========================================================================== 1832 // Canned table and table descriptor creation 1833 1834 public final static byte[] fam1 = Bytes.toBytes("colfamily11"); 1835 public final static byte[] fam2 = Bytes.toBytes("colfamily21"); 1836 public final static byte[] fam3 = Bytes.toBytes("colfamily31"); 1837 public static final byte[][] COLUMNS = { fam1, fam2, fam3 }; 1838 private static final int MAXVERSIONS = 3; 1839 1840 public static final char FIRST_CHAR = 'a'; 1841 public static final char LAST_CHAR = 'z'; 1842 public static final byte[] START_KEY_BYTES = { FIRST_CHAR, FIRST_CHAR, FIRST_CHAR }; 1843 public static final String START_KEY = new String(START_KEY_BYTES, HConstants.UTF8_CHARSET); 1844 1845 public TableDescriptorBuilder createModifyableTableDescriptor(final String name) { 1846 return createModifyableTableDescriptor(TableName.valueOf(name), 1847 ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, MAXVERSIONS, HConstants.FOREVER, 1848 ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED); 1849 } 1850 1851 public TableDescriptor createTableDescriptor(final TableName name, final int minVersions, 1852 final int versions, final int ttl, KeepDeletedCells keepDeleted) { 1853 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(name); 1854 for (byte[] cfName : new byte[][] { fam1, fam2, fam3 }) { 1855 ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(cfName) 1856 .setMinVersions(minVersions).setMaxVersions(versions).setKeepDeletedCells(keepDeleted) 1857 .setBlockCacheEnabled(false).setTimeToLive(ttl); 1858 if (isNewVersionBehaviorEnabled()) { 1859 cfBuilder.setNewVersionBehavior(true); 1860 } 1861 builder.setColumnFamily(cfBuilder.build()); 1862 } 1863 return builder.build(); 1864 } 1865 1866 public TableDescriptorBuilder createModifyableTableDescriptor(final TableName name, 1867 final int minVersions, final int versions, final int ttl, KeepDeletedCells keepDeleted) { 1868 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(name); 1869 for (byte[] cfName : new byte[][] { fam1, fam2, fam3 }) { 1870 ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(cfName) 1871 .setMinVersions(minVersions).setMaxVersions(versions).setKeepDeletedCells(keepDeleted) 1872 .setBlockCacheEnabled(false).setTimeToLive(ttl); 1873 if (isNewVersionBehaviorEnabled()) { 1874 cfBuilder.setNewVersionBehavior(true); 1875 } 1876 builder.setColumnFamily(cfBuilder.build()); 1877 } 1878 return builder; 1879 } 1880 1881 /** 1882 * Create a table of name <code>name</code>. 1883 * @param name Name to give table. 1884 * @return Column descriptor. 1885 */ 1886 public TableDescriptor createTableDescriptor(final TableName name) { 1887 return createTableDescriptor(name, ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 1888 MAXVERSIONS, HConstants.FOREVER, ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED); 1889 } 1890 1891 public TableDescriptor createTableDescriptor(final TableName tableName, byte[] family) { 1892 return createTableDescriptor(tableName, new byte[][] { family }, 1); 1893 } 1894 1895 public TableDescriptor createTableDescriptor(final TableName tableName, byte[][] families, 1896 int maxVersions) { 1897 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1898 for (byte[] family : families) { 1899 ColumnFamilyDescriptorBuilder cfBuilder = 1900 ColumnFamilyDescriptorBuilder.newBuilder(family).setMaxVersions(maxVersions); 1901 if (isNewVersionBehaviorEnabled()) { 1902 cfBuilder.setNewVersionBehavior(true); 1903 } 1904 builder.setColumnFamily(cfBuilder.build()); 1905 } 1906 return builder.build(); 1907 } 1908 1909 /** 1910 * Create an HRegion that writes to the local tmp dirs 1911 * @param desc a table descriptor indicating which table the region belongs to 1912 * @param startKey the start boundary of the region 1913 * @param endKey the end boundary of the region 1914 * @return a region that writes to local dir for testing 1915 */ 1916 public HRegion createLocalHRegion(TableDescriptor desc, byte[] startKey, byte[] endKey) 1917 throws IOException { 1918 RegionInfo hri = RegionInfoBuilder.newBuilder(desc.getTableName()).setStartKey(startKey) 1919 .setEndKey(endKey).build(); 1920 return createLocalHRegion(hri, desc); 1921 } 1922 1923 /** 1924 * Create an HRegion that writes to the local tmp dirs. Creates the WAL for you. Be sure to call 1925 * {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} when you're finished with it. 1926 */ 1927 public HRegion createLocalHRegion(RegionInfo info, TableDescriptor desc) throws IOException { 1928 return createRegionAndWAL(info, getDataTestDir(), getConfiguration(), desc); 1929 } 1930 1931 /** 1932 * Create an HRegion that writes to the local tmp dirs with specified wal 1933 * @param info regioninfo 1934 * @param conf configuration 1935 * @param desc table descriptor 1936 * @param wal wal for this region. 1937 * @return created hregion 1938 */ 1939 public HRegion createLocalHRegion(RegionInfo info, Configuration conf, TableDescriptor desc, 1940 WAL wal) throws IOException { 1941 return HRegion.createHRegion(info, getDataTestDir(), conf, desc, wal); 1942 } 1943 1944 /** 1945 * Return a region on which you must call {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} 1946 * when done. 1947 */ 1948 public HRegion createLocalHRegion(TableName tableName, byte[] startKey, byte[] stopKey, 1949 Configuration conf, boolean isReadOnly, Durability durability, WAL wal, byte[]... families) 1950 throws IOException { 1951 return createLocalHRegionWithInMemoryFlags(tableName, startKey, stopKey, conf, isReadOnly, 1952 durability, wal, null, families); 1953 } 1954 1955 public HRegion createLocalHRegionWithInMemoryFlags(TableName tableName, byte[] startKey, 1956 byte[] stopKey, Configuration conf, boolean isReadOnly, Durability durability, WAL wal, 1957 boolean[] compactedMemStore, byte[]... families) throws IOException { 1958 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 1959 builder.setReadOnly(isReadOnly); 1960 int i = 0; 1961 for (byte[] family : families) { 1962 ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(family); 1963 if (compactedMemStore != null && i < compactedMemStore.length) { 1964 cfBuilder.setInMemoryCompaction(MemoryCompactionPolicy.BASIC); 1965 } else { 1966 cfBuilder.setInMemoryCompaction(MemoryCompactionPolicy.NONE); 1967 1968 } 1969 i++; 1970 // Set default to be three versions. 1971 cfBuilder.setMaxVersions(Integer.MAX_VALUE); 1972 builder.setColumnFamily(cfBuilder.build()); 1973 } 1974 builder.setDurability(durability); 1975 RegionInfo info = 1976 RegionInfoBuilder.newBuilder(tableName).setStartKey(startKey).setEndKey(stopKey).build(); 1977 return createLocalHRegion(info, conf, builder.build(), wal); 1978 } 1979 1980 // 1981 // ========================================================================== 1982 1983 /** 1984 * Provide an existing table name to truncate. Scans the table and issues a delete for each row 1985 * read. 1986 * @param tableName existing table 1987 * @return HTable to that new table 1988 */ 1989 public Table deleteTableData(TableName tableName) throws IOException { 1990 Table table = getConnection().getTable(tableName); 1991 Scan scan = new Scan(); 1992 ResultScanner resScan = table.getScanner(scan); 1993 for (Result res : resScan) { 1994 Delete del = new Delete(res.getRow()); 1995 table.delete(del); 1996 } 1997 resScan = table.getScanner(scan); 1998 resScan.close(); 1999 return table; 2000 } 2001 2002 /** 2003 * Truncate a table using the admin command. Effectively disables, deletes, and recreates the 2004 * table. 2005 * @param tableName table which must exist. 2006 * @param preserveRegions keep the existing split points 2007 * @return HTable for the new table 2008 */ 2009 public Table truncateTable(final TableName tableName, final boolean preserveRegions) 2010 throws IOException { 2011 Admin admin = getAdmin(); 2012 if (!admin.isTableDisabled(tableName)) { 2013 admin.disableTable(tableName); 2014 } 2015 admin.truncateTable(tableName, preserveRegions); 2016 return getConnection().getTable(tableName); 2017 } 2018 2019 /** 2020 * Truncate a table using the admin command. Effectively disables, deletes, and recreates the 2021 * table. For previous behavior of issuing row deletes, see deleteTableData. Expressly does not 2022 * preserve regions of existing table. 2023 * @param tableName table which must exist. 2024 * @return HTable for the new table 2025 */ 2026 public Table truncateTable(final TableName tableName) throws IOException { 2027 return truncateTable(tableName, false); 2028 } 2029 2030 /** 2031 * Load table with rows from 'aaa' to 'zzz'. 2032 * @param t Table 2033 * @param f Family 2034 * @return Count of rows loaded. 2035 */ 2036 public int loadTable(final Table t, final byte[] f) throws IOException { 2037 return loadTable(t, new byte[][] { f }); 2038 } 2039 2040 /** 2041 * Load table with rows from 'aaa' to 'zzz'. 2042 * @param t Table 2043 * @param f Family 2044 * @return Count of rows loaded. 2045 */ 2046 public int loadTable(final Table t, final byte[] f, boolean writeToWAL) throws IOException { 2047 return loadTable(t, new byte[][] { f }, null, writeToWAL); 2048 } 2049 2050 /** 2051 * Load table of multiple column families with rows from 'aaa' to 'zzz'. 2052 * @param t Table 2053 * @param f Array of Families to load 2054 * @return Count of rows loaded. 2055 */ 2056 public int loadTable(final Table t, final byte[][] f) throws IOException { 2057 return loadTable(t, f, null); 2058 } 2059 2060 /** 2061 * Load table of multiple column families with rows from 'aaa' to 'zzz'. 2062 * @param t Table 2063 * @param f Array of Families to load 2064 * @param value the values of the cells. If null is passed, the row key is used as value 2065 * @return Count of rows loaded. 2066 */ 2067 public int loadTable(final Table t, final byte[][] f, byte[] value) throws IOException { 2068 return loadTable(t, f, value, true); 2069 } 2070 2071 /** 2072 * Load table of multiple column families with rows from 'aaa' to 'zzz'. 2073 * @param t Table 2074 * @param f Array of Families to load 2075 * @param value the values of the cells. If null is passed, the row key is used as value 2076 * @return Count of rows loaded. 2077 */ 2078 public int loadTable(final Table t, final byte[][] f, byte[] value, boolean writeToWAL) 2079 throws IOException { 2080 List<Put> puts = new ArrayList<>(); 2081 for (byte[] row : HBaseTestingUtility.ROWS) { 2082 Put put = new Put(row); 2083 put.setDurability(writeToWAL ? Durability.USE_DEFAULT : Durability.SKIP_WAL); 2084 for (int i = 0; i < f.length; i++) { 2085 byte[] value1 = value != null ? value : row; 2086 put.addColumn(f[i], f[i], value1); 2087 } 2088 puts.add(put); 2089 } 2090 t.put(puts); 2091 return puts.size(); 2092 } 2093 2094 /** 2095 * A tracker for tracking and validating table rows generated with 2096 * {@link HBaseTestingUtility#loadTable(Table, byte[])} 2097 */ 2098 public static class SeenRowTracker { 2099 int dim = 'z' - 'a' + 1; 2100 int[][][] seenRows = new int[dim][dim][dim]; // count of how many times the row is seen 2101 byte[] startRow; 2102 byte[] stopRow; 2103 2104 public SeenRowTracker(byte[] startRow, byte[] stopRow) { 2105 this.startRow = startRow; 2106 this.stopRow = stopRow; 2107 } 2108 2109 void reset() { 2110 for (byte[] row : ROWS) { 2111 seenRows[i(row[0])][i(row[1])][i(row[2])] = 0; 2112 } 2113 } 2114 2115 int i(byte b) { 2116 return b - 'a'; 2117 } 2118 2119 public void addRow(byte[] row) { 2120 seenRows[i(row[0])][i(row[1])][i(row[2])]++; 2121 } 2122 2123 /** 2124 * Validate that all the rows between startRow and stopRow are seen exactly once, and all other 2125 * rows none 2126 */ 2127 public void validate() { 2128 for (byte b1 = 'a'; b1 <= 'z'; b1++) { 2129 for (byte b2 = 'a'; b2 <= 'z'; b2++) { 2130 for (byte b3 = 'a'; b3 <= 'z'; b3++) { 2131 int count = seenRows[i(b1)][i(b2)][i(b3)]; 2132 int expectedCount = 0; 2133 if ( 2134 Bytes.compareTo(new byte[] { b1, b2, b3 }, startRow) >= 0 2135 && Bytes.compareTo(new byte[] { b1, b2, b3 }, stopRow) < 0 2136 ) { 2137 expectedCount = 1; 2138 } 2139 if (count != expectedCount) { 2140 String row = new String(new byte[] { b1, b2, b3 }, StandardCharsets.UTF_8); 2141 throw new RuntimeException("Row:" + row + " has a seen count of " + count + " " 2142 + "instead of " + expectedCount); 2143 } 2144 } 2145 } 2146 } 2147 } 2148 } 2149 2150 public int loadRegion(final HRegion r, final byte[] f) throws IOException { 2151 return loadRegion(r, f, false); 2152 } 2153 2154 public int loadRegion(final Region r, final byte[] f) throws IOException { 2155 return loadRegion((HRegion) r, f); 2156 } 2157 2158 /** 2159 * Load region with rows from 'aaa' to 'zzz'. 2160 * @param r Region 2161 * @param f Family 2162 * @param flush flush the cache if true 2163 * @return Count of rows loaded. 2164 */ 2165 public int loadRegion(final HRegion r, final byte[] f, final boolean flush) throws IOException { 2166 byte[] k = new byte[3]; 2167 int rowCount = 0; 2168 for (byte b1 = 'a'; b1 <= 'z'; b1++) { 2169 for (byte b2 = 'a'; b2 <= 'z'; b2++) { 2170 for (byte b3 = 'a'; b3 <= 'z'; b3++) { 2171 k[0] = b1; 2172 k[1] = b2; 2173 k[2] = b3; 2174 Put put = new Put(k); 2175 put.setDurability(Durability.SKIP_WAL); 2176 put.addColumn(f, null, k); 2177 if (r.getWAL() == null) { 2178 put.setDurability(Durability.SKIP_WAL); 2179 } 2180 int preRowCount = rowCount; 2181 int pause = 10; 2182 int maxPause = 1000; 2183 while (rowCount == preRowCount) { 2184 try { 2185 r.put(put); 2186 rowCount++; 2187 } catch (RegionTooBusyException e) { 2188 pause = (pause * 2 >= maxPause) ? maxPause : pause * 2; 2189 Threads.sleep(pause); 2190 } 2191 } 2192 } 2193 } 2194 if (flush) { 2195 r.flush(true); 2196 } 2197 } 2198 return rowCount; 2199 } 2200 2201 public void loadNumericRows(final Table t, final byte[] f, int startRow, int endRow) 2202 throws IOException { 2203 for (int i = startRow; i < endRow; i++) { 2204 byte[] data = Bytes.toBytes(String.valueOf(i)); 2205 Put put = new Put(data); 2206 put.addColumn(f, null, data); 2207 t.put(put); 2208 } 2209 } 2210 2211 public void loadRandomRows(final Table t, final byte[] f, int rowSize, int totalRows) 2212 throws IOException { 2213 byte[] row = new byte[rowSize]; 2214 for (int i = 0; i < totalRows; i++) { 2215 Bytes.random(row); 2216 Put put = new Put(row); 2217 put.addColumn(f, new byte[] { 0 }, new byte[] { 0 }); 2218 t.put(put); 2219 } 2220 } 2221 2222 public void verifyNumericRows(Table table, final byte[] f, int startRow, int endRow, 2223 int replicaId) throws IOException { 2224 for (int i = startRow; i < endRow; i++) { 2225 String failMsg = "Failed verification of row :" + i; 2226 byte[] data = Bytes.toBytes(String.valueOf(i)); 2227 Get get = new Get(data); 2228 get.setReplicaId(replicaId); 2229 get.setConsistency(Consistency.TIMELINE); 2230 Result result = table.get(get); 2231 if (!result.containsColumn(f, null)) { 2232 throw new AssertionError(failMsg); 2233 } 2234 assertEquals(failMsg, 1, result.getColumnCells(f, null).size()); 2235 Cell cell = result.getColumnLatestCell(f, null); 2236 if ( 2237 !Bytes.equals(data, 0, data.length, cell.getValueArray(), cell.getValueOffset(), 2238 cell.getValueLength()) 2239 ) { 2240 throw new AssertionError(failMsg); 2241 } 2242 } 2243 } 2244 2245 public void verifyNumericRows(Region region, final byte[] f, int startRow, int endRow) 2246 throws IOException { 2247 verifyNumericRows((HRegion) region, f, startRow, endRow); 2248 } 2249 2250 public void verifyNumericRows(HRegion region, final byte[] f, int startRow, int endRow) 2251 throws IOException { 2252 verifyNumericRows(region, f, startRow, endRow, true); 2253 } 2254 2255 public void verifyNumericRows(Region region, final byte[] f, int startRow, int endRow, 2256 final boolean present) throws IOException { 2257 verifyNumericRows((HRegion) region, f, startRow, endRow, present); 2258 } 2259 2260 public void verifyNumericRows(HRegion region, final byte[] f, int startRow, int endRow, 2261 final boolean present) throws IOException { 2262 for (int i = startRow; i < endRow; i++) { 2263 String failMsg = "Failed verification of row :" + i; 2264 byte[] data = Bytes.toBytes(String.valueOf(i)); 2265 Result result = region.get(new Get(data)); 2266 2267 boolean hasResult = result != null && !result.isEmpty(); 2268 if (present != hasResult) { 2269 throw new AssertionError( 2270 failMsg + result + " expected:<" + present + "> but was:<" + hasResult + ">"); 2271 } 2272 if (!present) continue; 2273 2274 if (!result.containsColumn(f, null)) { 2275 throw new AssertionError(failMsg); 2276 } 2277 assertEquals(failMsg, 1, result.getColumnCells(f, null).size()); 2278 Cell cell = result.getColumnLatestCell(f, null); 2279 if ( 2280 !Bytes.equals(data, 0, data.length, cell.getValueArray(), cell.getValueOffset(), 2281 cell.getValueLength()) 2282 ) { 2283 throw new AssertionError(failMsg); 2284 } 2285 } 2286 } 2287 2288 public void deleteNumericRows(final Table t, final byte[] f, int startRow, int endRow) 2289 throws IOException { 2290 for (int i = startRow; i < endRow; i++) { 2291 byte[] data = Bytes.toBytes(String.valueOf(i)); 2292 Delete delete = new Delete(data); 2293 delete.addFamily(f); 2294 t.delete(delete); 2295 } 2296 } 2297 2298 /** 2299 * Return the number of rows in the given table. 2300 * @param table to count rows 2301 * @return count of rows 2302 */ 2303 public static int countRows(final Table table) throws IOException { 2304 return countRows(table, new Scan()); 2305 } 2306 2307 public static int countRows(final Table table, final Scan scan) throws IOException { 2308 try (ResultScanner results = table.getScanner(scan)) { 2309 int count = 0; 2310 while (results.next() != null) { 2311 count++; 2312 } 2313 return count; 2314 } 2315 } 2316 2317 public int countRows(final Table table, final byte[]... families) throws IOException { 2318 Scan scan = new Scan(); 2319 for (byte[] family : families) { 2320 scan.addFamily(family); 2321 } 2322 return countRows(table, scan); 2323 } 2324 2325 /** 2326 * Return the number of rows in the given table. 2327 */ 2328 public int countRows(final TableName tableName) throws IOException { 2329 Table table = getConnection().getTable(tableName); 2330 try { 2331 return countRows(table); 2332 } finally { 2333 table.close(); 2334 } 2335 } 2336 2337 public int countRows(final Region region) throws IOException { 2338 return countRows(region, new Scan()); 2339 } 2340 2341 public int countRows(final Region region, final Scan scan) throws IOException { 2342 InternalScanner scanner = region.getScanner(scan); 2343 try { 2344 return countRows(scanner); 2345 } finally { 2346 scanner.close(); 2347 } 2348 } 2349 2350 public int countRows(final InternalScanner scanner) throws IOException { 2351 int scannedCount = 0; 2352 List<Cell> results = new ArrayList<>(); 2353 boolean hasMore = true; 2354 while (hasMore) { 2355 hasMore = scanner.next(results); 2356 scannedCount += results.size(); 2357 results.clear(); 2358 } 2359 return scannedCount; 2360 } 2361 2362 /** 2363 * Return an md5 digest of the entire contents of a table. 2364 */ 2365 public String checksumRows(final Table table) throws Exception { 2366 2367 Scan scan = new Scan(); 2368 ResultScanner results = table.getScanner(scan); 2369 MessageDigest digest = MessageDigest.getInstance("MD5"); 2370 for (Result res : results) { 2371 digest.update(res.getRow()); 2372 } 2373 results.close(); 2374 return digest.toString(); 2375 } 2376 2377 /** All the row values for the data loaded by {@link #loadTable(Table, byte[])} */ 2378 public static final byte[][] ROWS = new byte[(int) Math.pow('z' - 'a' + 1, 3)][3]; // ~52KB 2379 static { 2380 int i = 0; 2381 for (byte b1 = 'a'; b1 <= 'z'; b1++) { 2382 for (byte b2 = 'a'; b2 <= 'z'; b2++) { 2383 for (byte b3 = 'a'; b3 <= 'z'; b3++) { 2384 ROWS[i][0] = b1; 2385 ROWS[i][1] = b2; 2386 ROWS[i][2] = b3; 2387 i++; 2388 } 2389 } 2390 } 2391 } 2392 2393 public static final byte[][] KEYS = { HConstants.EMPTY_BYTE_ARRAY, Bytes.toBytes("bbb"), 2394 Bytes.toBytes("ccc"), Bytes.toBytes("ddd"), Bytes.toBytes("eee"), Bytes.toBytes("fff"), 2395 Bytes.toBytes("ggg"), Bytes.toBytes("hhh"), Bytes.toBytes("iii"), Bytes.toBytes("jjj"), 2396 Bytes.toBytes("kkk"), Bytes.toBytes("lll"), Bytes.toBytes("mmm"), Bytes.toBytes("nnn"), 2397 Bytes.toBytes("ooo"), Bytes.toBytes("ppp"), Bytes.toBytes("qqq"), Bytes.toBytes("rrr"), 2398 Bytes.toBytes("sss"), Bytes.toBytes("ttt"), Bytes.toBytes("uuu"), Bytes.toBytes("vvv"), 2399 Bytes.toBytes("www"), Bytes.toBytes("xxx"), Bytes.toBytes("yyy") }; 2400 2401 public static final byte[][] KEYS_FOR_HBA_CREATE_TABLE = { Bytes.toBytes("bbb"), 2402 Bytes.toBytes("ccc"), Bytes.toBytes("ddd"), Bytes.toBytes("eee"), Bytes.toBytes("fff"), 2403 Bytes.toBytes("ggg"), Bytes.toBytes("hhh"), Bytes.toBytes("iii"), Bytes.toBytes("jjj"), 2404 Bytes.toBytes("kkk"), Bytes.toBytes("lll"), Bytes.toBytes("mmm"), Bytes.toBytes("nnn"), 2405 Bytes.toBytes("ooo"), Bytes.toBytes("ppp"), Bytes.toBytes("qqq"), Bytes.toBytes("rrr"), 2406 Bytes.toBytes("sss"), Bytes.toBytes("ttt"), Bytes.toBytes("uuu"), Bytes.toBytes("vvv"), 2407 Bytes.toBytes("www"), Bytes.toBytes("xxx"), Bytes.toBytes("yyy"), Bytes.toBytes("zzz") }; 2408 2409 /** 2410 * Create rows in hbase:meta for regions of the specified table with the specified start keys. The 2411 * first startKey should be a 0 length byte array if you want to form a proper range of regions. 2412 * @return list of region info for regions added to meta 2413 */ 2414 public List<RegionInfo> createMultiRegionsInMeta(final Configuration conf, 2415 final TableDescriptor htd, byte[][] startKeys) throws IOException { 2416 Table meta = getConnection().getTable(TableName.META_TABLE_NAME); 2417 Arrays.sort(startKeys, Bytes.BYTES_COMPARATOR); 2418 List<RegionInfo> newRegions = new ArrayList<>(startKeys.length); 2419 MetaTableAccessor.updateTableState(getConnection(), htd.getTableName(), 2420 TableState.State.ENABLED); 2421 // add custom ones 2422 for (int i = 0; i < startKeys.length; i++) { 2423 int j = (i + 1) % startKeys.length; 2424 RegionInfo hri = RegionInfoBuilder.newBuilder(htd.getTableName()).setStartKey(startKeys[i]) 2425 .setEndKey(startKeys[j]).build(); 2426 MetaTableAccessor.addRegionsToMeta(getConnection(), Collections.singletonList(hri), 1); 2427 newRegions.add(hri); 2428 } 2429 2430 meta.close(); 2431 return newRegions; 2432 } 2433 2434 /** 2435 * Create an unmanaged WAL. Be sure to close it when you're through. 2436 */ 2437 public static WAL createWal(final Configuration conf, final Path rootDir, final RegionInfo hri) 2438 throws IOException { 2439 // The WAL subsystem will use the default rootDir rather than the passed in rootDir 2440 // unless I pass along via the conf. 2441 Configuration confForWAL = new Configuration(conf); 2442 confForWAL.set(HConstants.HBASE_DIR, rootDir.toString()); 2443 return new WALFactory(confForWAL, "hregion-" + RandomStringUtils.randomNumeric(8)).getWAL(hri); 2444 } 2445 2446 /** 2447 * Create a region with it's own WAL. Be sure to call 2448 * {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} to clean up all resources. 2449 */ 2450 public static HRegion createRegionAndWAL(final RegionInfo info, final Path rootDir, 2451 final Configuration conf, final TableDescriptor htd) throws IOException { 2452 return createRegionAndWAL(info, rootDir, conf, htd, true); 2453 } 2454 2455 /** 2456 * Create a region with it's own WAL. Be sure to call 2457 * {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} to clean up all resources. 2458 */ 2459 public static HRegion createRegionAndWAL(final RegionInfo info, final Path rootDir, 2460 final Configuration conf, final TableDescriptor htd, BlockCache blockCache) throws IOException { 2461 HRegion region = createRegionAndWAL(info, rootDir, conf, htd, false); 2462 region.setBlockCache(blockCache); 2463 region.initialize(); 2464 return region; 2465 } 2466 2467 /** 2468 * Create a region with it's own WAL. Be sure to call 2469 * {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} to clean up all resources. 2470 */ 2471 public static HRegion createRegionAndWAL(final RegionInfo info, final Path rootDir, 2472 final Configuration conf, final TableDescriptor htd, MobFileCache mobFileCache) 2473 throws IOException { 2474 HRegion region = createRegionAndWAL(info, rootDir, conf, htd, false); 2475 region.setMobFileCache(mobFileCache); 2476 region.initialize(); 2477 return region; 2478 } 2479 2480 /** 2481 * Create a region with it's own WAL. Be sure to call 2482 * {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} to clean up all resources. 2483 */ 2484 public static HRegion createRegionAndWAL(final RegionInfo info, final Path rootDir, 2485 final Configuration conf, final TableDescriptor htd, boolean initialize) throws IOException { 2486 ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0, 0, null, 2487 MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT); 2488 WAL wal = createWal(conf, rootDir, info); 2489 return HRegion.createHRegion(info, rootDir, conf, htd, wal, initialize); 2490 } 2491 2492 /** 2493 * Returns all rows from the hbase:meta table. 2494 * @throws IOException When reading the rows fails. 2495 */ 2496 public List<byte[]> getMetaTableRows() throws IOException { 2497 // TODO: Redo using MetaTableAccessor class 2498 Table t = getConnection().getTable(TableName.META_TABLE_NAME); 2499 List<byte[]> rows = new ArrayList<>(); 2500 ResultScanner s = t.getScanner(new Scan()); 2501 for (Result result : s) { 2502 LOG.info("getMetaTableRows: row -> " + Bytes.toStringBinary(result.getRow())); 2503 rows.add(result.getRow()); 2504 } 2505 s.close(); 2506 t.close(); 2507 return rows; 2508 } 2509 2510 /** 2511 * Returns all rows from the hbase:meta table for a given user table 2512 * @throws IOException When reading the rows fails. 2513 */ 2514 public List<byte[]> getMetaTableRows(TableName tableName) throws IOException { 2515 // TODO: Redo using MetaTableAccessor. 2516 Table t = getConnection().getTable(TableName.META_TABLE_NAME); 2517 List<byte[]> rows = new ArrayList<>(); 2518 ResultScanner s = t.getScanner(new Scan()); 2519 for (Result result : s) { 2520 RegionInfo info = CatalogFamilyFormat.getRegionInfo(result); 2521 if (info == null) { 2522 LOG.error("No region info for row " + Bytes.toString(result.getRow())); 2523 // TODO figure out what to do for this new hosed case. 2524 continue; 2525 } 2526 2527 if (info.getTable().equals(tableName)) { 2528 LOG.info("getMetaTableRows: row -> " + Bytes.toStringBinary(result.getRow()) + info); 2529 rows.add(result.getRow()); 2530 } 2531 } 2532 s.close(); 2533 t.close(); 2534 return rows; 2535 } 2536 2537 /** 2538 * Returns all regions of the specified table 2539 * @param tableName the table name 2540 * @return all regions of the specified table 2541 * @throws IOException when getting the regions fails. 2542 */ 2543 private List<RegionInfo> getRegions(TableName tableName) throws IOException { 2544 try (Admin admin = getConnection().getAdmin()) { 2545 return admin.getRegions(tableName); 2546 } 2547 } 2548 2549 /** 2550 * Find any other region server which is different from the one identified by parameter 2551 * @return another region server 2552 */ 2553 public HRegionServer getOtherRegionServer(HRegionServer rs) { 2554 for (JVMClusterUtil.RegionServerThread rst : getMiniHBaseCluster().getRegionServerThreads()) { 2555 if (!(rst.getRegionServer() == rs)) { 2556 return rst.getRegionServer(); 2557 } 2558 } 2559 return null; 2560 } 2561 2562 /** 2563 * Tool to get the reference to the region server object that holds the region of the specified 2564 * user table. 2565 * @param tableName user table to lookup in hbase:meta 2566 * @return region server that holds it, null if the row doesn't exist 2567 */ 2568 public HRegionServer getRSForFirstRegionInTable(TableName tableName) 2569 throws IOException, InterruptedException { 2570 List<RegionInfo> regions = getRegions(tableName); 2571 if (regions == null || regions.isEmpty()) { 2572 return null; 2573 } 2574 LOG.debug("Found " + regions.size() + " regions for table " + tableName); 2575 2576 byte[] firstRegionName = 2577 regions.stream().filter(r -> !r.isOffline()).map(RegionInfo::getRegionName).findFirst() 2578 .orElseThrow(() -> new IOException("online regions not found in table " + tableName)); 2579 2580 LOG.debug("firstRegionName=" + Bytes.toString(firstRegionName)); 2581 long pause = getConfiguration().getLong(HConstants.HBASE_CLIENT_PAUSE, 2582 HConstants.DEFAULT_HBASE_CLIENT_PAUSE); 2583 int numRetries = getConfiguration().getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2584 HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER); 2585 RetryCounter retrier = new RetryCounter(numRetries + 1, (int) pause, TimeUnit.MICROSECONDS); 2586 while (retrier.shouldRetry()) { 2587 int index = getMiniHBaseCluster().getServerWith(firstRegionName); 2588 if (index != -1) { 2589 return getMiniHBaseCluster().getRegionServerThreads().get(index).getRegionServer(); 2590 } 2591 // Came back -1. Region may not be online yet. Sleep a while. 2592 retrier.sleepUntilNextRetry(); 2593 } 2594 return null; 2595 } 2596 2597 /** 2598 * Starts a <code>MiniMRCluster</code> with a default number of <code>TaskTracker</code>'s. 2599 * @throws IOException When starting the cluster fails. 2600 */ 2601 public MiniMRCluster startMiniMapReduceCluster() throws IOException { 2602 // Set a very high max-disk-utilization percentage to avoid the NodeManagers from failing. 2603 conf.setIfUnset("yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage", 2604 "99.0"); 2605 startMiniMapReduceCluster(2); 2606 return mrCluster; 2607 } 2608 2609 /** 2610 * Tasktracker has a bug where changing the hadoop.log.dir system property will not change its 2611 * internal static LOG_DIR variable. 2612 */ 2613 private void forceChangeTaskLogDir() { 2614 Field logDirField; 2615 try { 2616 logDirField = TaskLog.class.getDeclaredField("LOG_DIR"); 2617 logDirField.setAccessible(true); 2618 2619 Field modifiersField = ReflectionUtils.getModifiersField(); 2620 modifiersField.setAccessible(true); 2621 modifiersField.setInt(logDirField, logDirField.getModifiers() & ~Modifier.FINAL); 2622 2623 logDirField.set(null, new File(hadoopLogDir, "userlogs")); 2624 } catch (SecurityException e) { 2625 throw new RuntimeException(e); 2626 } catch (NoSuchFieldException e) { 2627 throw new RuntimeException(e); 2628 } catch (IllegalArgumentException e) { 2629 throw new RuntimeException(e); 2630 } catch (IllegalAccessException e) { 2631 throw new RuntimeException(e); 2632 } 2633 } 2634 2635 /** 2636 * Starts a <code>MiniMRCluster</code>. Call {@link #setFileSystemURI(String)} to use a different 2637 * filesystem. 2638 * @param servers The number of <code>TaskTracker</code>'s to start. 2639 * @throws IOException When starting the cluster fails. 2640 */ 2641 private void startMiniMapReduceCluster(final int servers) throws IOException { 2642 if (mrCluster != null) { 2643 throw new IllegalStateException("MiniMRCluster is already running"); 2644 } 2645 LOG.info("Starting mini mapreduce cluster..."); 2646 setupClusterTestDir(); 2647 createDirsAndSetProperties(); 2648 2649 forceChangeTaskLogDir(); 2650 2651 //// hadoop2 specific settings 2652 // Tests were failing because this process used 6GB of virtual memory and was getting killed. 2653 // we up the VM usable so that processes don't get killed. 2654 conf.setFloat("yarn.nodemanager.vmem-pmem-ratio", 8.0f); 2655 2656 // Tests were failing due to MAPREDUCE-4880 / MAPREDUCE-4607 against hadoop 2.0.2-alpha and 2657 // this avoids the problem by disabling speculative task execution in tests. 2658 conf.setBoolean("mapreduce.map.speculative", false); 2659 conf.setBoolean("mapreduce.reduce.speculative", false); 2660 //// 2661 2662 // Allow the user to override FS URI for this map-reduce cluster to use. 2663 mrCluster = 2664 new MiniMRCluster(servers, FS_URI != null ? FS_URI : FileSystem.get(conf).getUri().toString(), 2665 1, null, null, new JobConf(this.conf)); 2666 JobConf jobConf = MapreduceTestingShim.getJobConf(mrCluster); 2667 if (jobConf == null) { 2668 jobConf = mrCluster.createJobConf(); 2669 } 2670 // Hadoop MiniMR overwrites this while it should not 2671 jobConf.set("mapreduce.cluster.local.dir", conf.get("mapreduce.cluster.local.dir")); 2672 LOG.info("Mini mapreduce cluster started"); 2673 2674 // In hadoop2, YARN/MR2 starts a mini cluster with its own conf instance and updates settings. 2675 // Our HBase MR jobs need several of these settings in order to properly run. So we copy the 2676 // necessary config properties here. YARN-129 required adding a few properties. 2677 conf.set("mapreduce.jobtracker.address", jobConf.get("mapreduce.jobtracker.address")); 2678 // this for mrv2 support; mr1 ignores this 2679 conf.set("mapreduce.framework.name", "yarn"); 2680 conf.setBoolean("yarn.is.minicluster", true); 2681 String rmAddress = jobConf.get("yarn.resourcemanager.address"); 2682 if (rmAddress != null) { 2683 conf.set("yarn.resourcemanager.address", rmAddress); 2684 } 2685 String historyAddress = jobConf.get("mapreduce.jobhistory.address"); 2686 if (historyAddress != null) { 2687 conf.set("mapreduce.jobhistory.address", historyAddress); 2688 } 2689 String schedulerAddress = jobConf.get("yarn.resourcemanager.scheduler.address"); 2690 if (schedulerAddress != null) { 2691 conf.set("yarn.resourcemanager.scheduler.address", schedulerAddress); 2692 } 2693 String mrJobHistoryWebappAddress = jobConf.get("mapreduce.jobhistory.webapp.address"); 2694 if (mrJobHistoryWebappAddress != null) { 2695 conf.set("mapreduce.jobhistory.webapp.address", mrJobHistoryWebappAddress); 2696 } 2697 String yarnRMWebappAddress = jobConf.get("yarn.resourcemanager.webapp.address"); 2698 if (yarnRMWebappAddress != null) { 2699 conf.set("yarn.resourcemanager.webapp.address", yarnRMWebappAddress); 2700 } 2701 } 2702 2703 /** 2704 * Stops the previously started <code>MiniMRCluster</code>. 2705 */ 2706 public void shutdownMiniMapReduceCluster() { 2707 if (mrCluster != null) { 2708 LOG.info("Stopping mini mapreduce cluster..."); 2709 mrCluster.shutdown(); 2710 mrCluster = null; 2711 LOG.info("Mini mapreduce cluster stopped"); 2712 } 2713 // Restore configuration to point to local jobtracker 2714 conf.set("mapreduce.jobtracker.address", "local"); 2715 } 2716 2717 /** 2718 * Create a stubbed out RegionServerService, mainly for getting FS. 2719 */ 2720 public RegionServerServices createMockRegionServerService() throws IOException { 2721 return createMockRegionServerService((ServerName) null); 2722 } 2723 2724 /** 2725 * Create a stubbed out RegionServerService, mainly for getting FS. This version is used by 2726 * TestTokenAuthentication 2727 */ 2728 public RegionServerServices createMockRegionServerService(RpcServerInterface rpc) 2729 throws IOException { 2730 final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher()); 2731 rss.setFileSystem(getTestFileSystem()); 2732 rss.setRpcServer(rpc); 2733 return rss; 2734 } 2735 2736 /** 2737 * Create a stubbed out RegionServerService, mainly for getting FS. This version is used by 2738 * TestOpenRegionHandler 2739 */ 2740 public RegionServerServices createMockRegionServerService(ServerName name) throws IOException { 2741 final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher(), name); 2742 rss.setFileSystem(getTestFileSystem()); 2743 return rss; 2744 } 2745 2746 /** 2747 * Switches the logger for the given class to DEBUG level. 2748 * @param clazz The class for which to switch to debug logging. 2749 * @deprecated In 2.3.0, will be removed in 4.0.0. Only support changing log level on log4j now as 2750 * HBase only uses log4j. You should do this by your own as it you know which log 2751 * framework you are using then set the log level to debug is very easy. 2752 */ 2753 @Deprecated 2754 public void enableDebug(Class<?> clazz) { 2755 Log4jUtils.enableDebug(clazz); 2756 } 2757 2758 /** 2759 * Expire the Master's session 2760 */ 2761 public void expireMasterSession() throws Exception { 2762 HMaster master = getMiniHBaseCluster().getMaster(); 2763 expireSession(master.getZooKeeper(), false); 2764 } 2765 2766 /** 2767 * Expire a region server's session 2768 * @param index which RS 2769 */ 2770 public void expireRegionServerSession(int index) throws Exception { 2771 HRegionServer rs = getMiniHBaseCluster().getRegionServer(index); 2772 expireSession(rs.getZooKeeper(), false); 2773 decrementMinRegionServerCount(); 2774 } 2775 2776 private void decrementMinRegionServerCount() { 2777 // decrement the count for this.conf, for newly spwaned master 2778 // this.hbaseCluster shares this configuration too 2779 decrementMinRegionServerCount(getConfiguration()); 2780 2781 // each master thread keeps a copy of configuration 2782 for (MasterThread master : getHBaseCluster().getMasterThreads()) { 2783 decrementMinRegionServerCount(master.getMaster().getConfiguration()); 2784 } 2785 } 2786 2787 private void decrementMinRegionServerCount(Configuration conf) { 2788 int currentCount = conf.getInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, -1); 2789 if (currentCount != -1) { 2790 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, Math.max(currentCount - 1, 1)); 2791 } 2792 } 2793 2794 public void expireSession(ZKWatcher nodeZK) throws Exception { 2795 expireSession(nodeZK, false); 2796 } 2797 2798 /** 2799 * Expire a ZooKeeper session as recommended in ZooKeeper documentation <a href= 2800 * "https://hbase.apache.org/docs/troubleshooting#troubleshooting-zookeeper">Troubleshooting 2801 * ZooKeeper</a> 2802 * <p/> 2803 * There are issues when doing this: 2804 * <ol> 2805 * <li>http://www.mail-archive.com/dev@zookeeper.apache.org/msg01942.html</li> 2806 * <li>https://issues.apache.org/jira/browse/ZOOKEEPER-1105</li> 2807 * </ol> 2808 * @param nodeZK - the ZK watcher to expire 2809 * @param checkStatus - true to check if we can create a Table with the current configuration. 2810 */ 2811 public void expireSession(ZKWatcher nodeZK, boolean checkStatus) throws Exception { 2812 Configuration c = new Configuration(this.conf); 2813 String quorumServers = ZKConfig.getZKQuorumServersString(c); 2814 ZooKeeper zk = nodeZK.getRecoverableZooKeeper().getZooKeeper(); 2815 byte[] password = zk.getSessionPasswd(); 2816 long sessionID = zk.getSessionId(); 2817 2818 // Expiry seems to be asynchronous (see comment from P. Hunt in [1]), 2819 // so we create a first watcher to be sure that the 2820 // event was sent. We expect that if our watcher receives the event 2821 // other watchers on the same machine will get is as well. 2822 // When we ask to close the connection, ZK does not close it before 2823 // we receive all the events, so don't have to capture the event, just 2824 // closing the connection should be enough. 2825 ZooKeeper monitor = new ZooKeeper(quorumServers, 1000, new org.apache.zookeeper.Watcher() { 2826 @Override 2827 public void process(WatchedEvent watchedEvent) { 2828 LOG.info("Monitor ZKW received event=" + watchedEvent); 2829 } 2830 }, sessionID, password); 2831 2832 // Making it expire 2833 ZooKeeper newZK = 2834 new ZooKeeper(quorumServers, 1000, EmptyWatcher.instance, sessionID, password); 2835 2836 // ensure that we have connection to the server before closing down, otherwise 2837 // the close session event will be eaten out before we start CONNECTING state 2838 long start = EnvironmentEdgeManager.currentTime(); 2839 while ( 2840 newZK.getState() != States.CONNECTED && EnvironmentEdgeManager.currentTime() - start < 1000 2841 ) { 2842 Thread.sleep(1); 2843 } 2844 newZK.close(); 2845 LOG.info("ZK Closed Session 0x" + Long.toHexString(sessionID)); 2846 2847 // Now closing & waiting to be sure that the clients get it. 2848 monitor.close(); 2849 2850 if (checkStatus) { 2851 getConnection().getTable(TableName.META_TABLE_NAME).close(); 2852 } 2853 } 2854 2855 /** 2856 * Get the Mini HBase cluster. 2857 * @return hbase cluster 2858 * @see #getHBaseClusterInterface() 2859 */ 2860 public MiniHBaseCluster getHBaseCluster() { 2861 return getMiniHBaseCluster(); 2862 } 2863 2864 /** 2865 * Returns the HBaseCluster instance. 2866 * <p> 2867 * Returned object can be any of the subclasses of HBaseCluster, and the tests referring this 2868 * should not assume that the cluster is a mini cluster or a distributed one. If the test only 2869 * works on a mini cluster, then specific method {@link #getMiniHBaseCluster()} can be used 2870 * instead w/o the need to type-cast. 2871 */ 2872 public HBaseCluster getHBaseClusterInterface() { 2873 // implementation note: we should rename this method as #getHBaseCluster(), 2874 // but this would require refactoring 90+ calls. 2875 return hbaseCluster; 2876 } 2877 2878 /** 2879 * Resets the connections so that the next time getConnection() is called, a new connection is 2880 * created. This is needed in cases where the entire cluster / all the masters are shutdown and 2881 * the connection is not valid anymore. TODO: There should be a more coherent way of doing this. 2882 * Unfortunately the way tests are written, not all start() stop() calls go through this class. 2883 * Most tests directly operate on the underlying mini/local hbase cluster. That makes it difficult 2884 * for this wrapper class to maintain the connection state automatically. Cleaning this is a much 2885 * bigger refactor. 2886 */ 2887 public void invalidateConnection() throws IOException { 2888 closeConnection(); 2889 // Update the master addresses if they changed. 2890 final String masterConfigBefore = conf.get(HConstants.MASTER_ADDRS_KEY); 2891 final String masterConfAfter = getMiniHBaseCluster().conf.get(HConstants.MASTER_ADDRS_KEY); 2892 LOG.info("Invalidated connection. Updating master addresses before: {} after: {}", 2893 masterConfigBefore, masterConfAfter); 2894 conf.set(HConstants.MASTER_ADDRS_KEY, 2895 getMiniHBaseCluster().conf.get(HConstants.MASTER_ADDRS_KEY)); 2896 } 2897 2898 /** 2899 * Get a shared Connection to the cluster. this method is thread safe. 2900 * @return A Connection that can be shared. Don't close. Will be closed on shutdown of cluster. 2901 */ 2902 public Connection getConnection() throws IOException { 2903 return getAsyncConnection().toConnection(); 2904 } 2905 2906 /** 2907 * Get a assigned Connection to the cluster. this method is thread safe. 2908 * @param user assigned user 2909 * @return A Connection with assigned user. 2910 */ 2911 public Connection getConnection(User user) throws IOException { 2912 return getAsyncConnection(user).toConnection(); 2913 } 2914 2915 /** 2916 * Get a shared AsyncClusterConnection to the cluster. this method is thread safe. 2917 * @return An AsyncClusterConnection that can be shared. Don't close. Will be closed on shutdown 2918 * of cluster. 2919 */ 2920 public AsyncClusterConnection getAsyncConnection() throws IOException { 2921 try { 2922 return asyncConnection.updateAndGet(connection -> { 2923 if (connection == null) { 2924 try { 2925 User user = UserProvider.instantiate(conf).getCurrent(); 2926 connection = getAsyncConnection(user); 2927 } catch (IOException ioe) { 2928 throw new UncheckedIOException("Failed to create connection", ioe); 2929 } 2930 } 2931 return connection; 2932 }); 2933 } catch (UncheckedIOException exception) { 2934 throw exception.getCause(); 2935 } 2936 } 2937 2938 /** 2939 * Get a assigned AsyncClusterConnection to the cluster. this method is thread safe. 2940 * @param user assigned user 2941 * @return An AsyncClusterConnection with assigned user. 2942 */ 2943 public AsyncClusterConnection getAsyncConnection(User user) throws IOException { 2944 return ClusterConnectionFactory.createAsyncClusterConnection(conf, null, user); 2945 } 2946 2947 public void closeConnection() throws IOException { 2948 if (hbaseAdmin != null) { 2949 Closeables.close(hbaseAdmin, true); 2950 hbaseAdmin = null; 2951 } 2952 AsyncClusterConnection asyncConnection = this.asyncConnection.getAndSet(null); 2953 if (asyncConnection != null) { 2954 Closeables.close(asyncConnection, true); 2955 } 2956 } 2957 2958 /** 2959 * Returns an Admin instance which is shared between HBaseTestingUtility instance users. Closing 2960 * it has no effect, it will be closed automatically when the cluster shutdowns 2961 */ 2962 public Admin getAdmin() throws IOException { 2963 if (hbaseAdmin == null) { 2964 this.hbaseAdmin = getConnection().getAdmin(); 2965 } 2966 return hbaseAdmin; 2967 } 2968 2969 private Admin hbaseAdmin = null; 2970 2971 /** 2972 * Returns an {@link Hbck} instance. Needs be closed when done. 2973 */ 2974 public Hbck getHbck() throws IOException { 2975 return getConnection().getHbck(); 2976 } 2977 2978 /** 2979 * Unassign the named region. 2980 * @param regionName The region to unassign. 2981 */ 2982 public void unassignRegion(String regionName) throws IOException { 2983 unassignRegion(Bytes.toBytes(regionName)); 2984 } 2985 2986 /** 2987 * Unassign the named region. 2988 * @param regionName The region to unassign. 2989 */ 2990 public void unassignRegion(byte[] regionName) throws IOException { 2991 getAdmin().unassign(regionName, true); 2992 } 2993 2994 /** 2995 * Closes the region containing the given row. 2996 * @param row The row to find the containing region. 2997 * @param table The table to find the region. 2998 */ 2999 public void unassignRegionByRow(String row, RegionLocator table) throws IOException { 3000 unassignRegionByRow(Bytes.toBytes(row), table); 3001 } 3002 3003 /** 3004 * Closes the region containing the given row. 3005 * @param row The row to find the containing region. 3006 * @param table The table to find the region. 3007 */ 3008 public void unassignRegionByRow(byte[] row, RegionLocator table) throws IOException { 3009 HRegionLocation hrl = table.getRegionLocation(row); 3010 unassignRegion(hrl.getRegion().getRegionName()); 3011 } 3012 3013 /** 3014 * Retrieves a splittable region randomly from tableName 3015 * @param tableName name of table 3016 * @param maxAttempts maximum number of attempts, unlimited for value of -1 3017 * @return the HRegion chosen, null if none was found within limit of maxAttempts 3018 */ 3019 public HRegion getSplittableRegion(TableName tableName, int maxAttempts) { 3020 List<HRegion> regions = getHBaseCluster().getRegions(tableName); 3021 int regCount = regions.size(); 3022 Set<Integer> attempted = new HashSet<>(); 3023 int idx; 3024 int attempts = 0; 3025 do { 3026 regions = getHBaseCluster().getRegions(tableName); 3027 if (regCount != regions.size()) { 3028 // if there was region movement, clear attempted Set 3029 attempted.clear(); 3030 } 3031 regCount = regions.size(); 3032 // There are chances that before we get the region for the table from an RS the region may 3033 // be going for CLOSE. This may be because online schema change is enabled 3034 if (regCount > 0) { 3035 idx = ThreadLocalRandom.current().nextInt(regCount); 3036 // if we have just tried this region, there is no need to try again 3037 if (attempted.contains(idx)) { 3038 continue; 3039 } 3040 HRegion region = regions.get(idx); 3041 if (region.checkSplit().isPresent()) { 3042 return region; 3043 } 3044 attempted.add(idx); 3045 } 3046 attempts++; 3047 } while (maxAttempts == -1 || attempts < maxAttempts); 3048 return null; 3049 } 3050 3051 public MiniDFSCluster getDFSCluster() { 3052 return dfsCluster; 3053 } 3054 3055 public void setDFSCluster(MiniDFSCluster cluster) throws IllegalStateException, IOException { 3056 setDFSCluster(cluster, true); 3057 } 3058 3059 /** 3060 * Set the MiniDFSCluster 3061 * @param cluster cluster to use 3062 * @param requireDown require the that cluster not be "up" (MiniDFSCluster#isClusterUp) before it 3063 * is set. 3064 * @throws IllegalStateException if the passed cluster is up when it is required to be down 3065 * @throws IOException if the FileSystem could not be set from the passed dfs cluster 3066 */ 3067 public void setDFSCluster(MiniDFSCluster cluster, boolean requireDown) 3068 throws IllegalStateException, IOException { 3069 if (dfsCluster != null && requireDown && dfsCluster.isClusterUp()) { 3070 throw new IllegalStateException("DFSCluster is already running! Shut it down first."); 3071 } 3072 this.dfsCluster = cluster; 3073 this.setFs(); 3074 } 3075 3076 public FileSystem getTestFileSystem() throws IOException { 3077 return HFileSystem.get(conf); 3078 } 3079 3080 /** 3081 * Wait until all regions in a table have been assigned. Waits default timeout before giving up 3082 * (30 seconds). 3083 * @param table Table to wait on. 3084 */ 3085 public void waitTableAvailable(TableName table) throws InterruptedException, IOException { 3086 waitTableAvailable(table.getName(), 30000); 3087 } 3088 3089 public void waitTableAvailable(TableName table, long timeoutMillis) 3090 throws InterruptedException, IOException { 3091 waitFor(timeoutMillis, predicateTableAvailable(table)); 3092 } 3093 3094 /** 3095 * Wait until all regions in a table have been assigned 3096 * @param table Table to wait on. 3097 * @param timeoutMillis Timeout. 3098 */ 3099 public void waitTableAvailable(byte[] table, long timeoutMillis) 3100 throws InterruptedException, IOException { 3101 waitFor(timeoutMillis, predicateTableAvailable(TableName.valueOf(table))); 3102 } 3103 3104 public String explainTableAvailability(TableName tableName) throws IOException { 3105 StringBuilder msg = 3106 new StringBuilder(explainTableState(tableName, TableState.State.ENABLED)).append(", "); 3107 if (getHBaseCluster().getMaster().isAlive()) { 3108 Map<RegionInfo, ServerName> assignments = getHBaseCluster().getMaster().getAssignmentManager() 3109 .getRegionStates().getRegionAssignments(); 3110 final List<Pair<RegionInfo, ServerName>> metaLocations = 3111 MetaTableAccessor.getTableRegionsAndLocations(getConnection(), tableName); 3112 for (Pair<RegionInfo, ServerName> metaLocation : metaLocations) { 3113 RegionInfo hri = metaLocation.getFirst(); 3114 ServerName sn = metaLocation.getSecond(); 3115 if (!assignments.containsKey(hri)) { 3116 msg.append(", region ").append(hri) 3117 .append(" not assigned, but found in meta, it expected to be on ").append(sn); 3118 } else if (sn == null) { 3119 msg.append(", region ").append(hri).append(" assigned, but has no server in meta"); 3120 } else if (!sn.equals(assignments.get(hri))) { 3121 msg.append(", region ").append(hri) 3122 .append(" assigned, but has different servers in meta and AM ( ").append(sn) 3123 .append(" <> ").append(assignments.get(hri)); 3124 } 3125 } 3126 } 3127 return msg.toString(); 3128 } 3129 3130 public String explainTableState(final TableName table, TableState.State state) 3131 throws IOException { 3132 TableState tableState = MetaTableAccessor.getTableState(getConnection(), table); 3133 if (tableState == null) { 3134 return "TableState in META: No table state in META for table " + table 3135 + " last state in meta (including deleted is " + findLastTableState(table) + ")"; 3136 } else if (!tableState.inStates(state)) { 3137 return "TableState in META: Not " + state + " state, but " + tableState; 3138 } else { 3139 return "TableState in META: OK"; 3140 } 3141 } 3142 3143 public TableState findLastTableState(final TableName table) throws IOException { 3144 final AtomicReference<TableState> lastTableState = new AtomicReference<>(null); 3145 ClientMetaTableAccessor.Visitor visitor = new ClientMetaTableAccessor.Visitor() { 3146 @Override 3147 public boolean visit(Result r) throws IOException { 3148 if (!Arrays.equals(r.getRow(), table.getName())) { 3149 return false; 3150 } 3151 TableState state = CatalogFamilyFormat.getTableState(r); 3152 if (state != null) { 3153 lastTableState.set(state); 3154 } 3155 return true; 3156 } 3157 }; 3158 MetaTableAccessor.scanMeta(getConnection(), null, null, ClientMetaTableAccessor.QueryType.TABLE, 3159 Integer.MAX_VALUE, visitor); 3160 return lastTableState.get(); 3161 } 3162 3163 /** 3164 * Waits for a table to be 'enabled'. Enabled means that table is set as 'enabled' and the regions 3165 * have been all assigned. Will timeout after default period (30 seconds) Tolerates nonexistent 3166 * table. 3167 * @param table the table to wait on. 3168 * @throws InterruptedException if interrupted while waiting 3169 * @throws IOException if an IO problem is encountered 3170 */ 3171 public void waitTableEnabled(TableName table) throws InterruptedException, IOException { 3172 waitTableEnabled(table, 30000); 3173 } 3174 3175 /** 3176 * Waits for a table to be 'enabled'. Enabled means that table is set as 'enabled' and the regions 3177 * have been all assigned. 3178 * @see #waitTableEnabled(TableName, long) 3179 * @param table Table to wait on. 3180 * @param timeoutMillis Time to wait on it being marked enabled. 3181 */ 3182 public void waitTableEnabled(byte[] table, long timeoutMillis) 3183 throws InterruptedException, IOException { 3184 waitTableEnabled(TableName.valueOf(table), timeoutMillis); 3185 } 3186 3187 public void waitTableEnabled(TableName table, long timeoutMillis) throws IOException { 3188 waitFor(timeoutMillis, predicateTableEnabled(table)); 3189 } 3190 3191 /** 3192 * Waits for a table to be 'disabled'. Disabled means that table is set as 'disabled' Will timeout 3193 * after default period (30 seconds) 3194 * @param table Table to wait on. 3195 */ 3196 public void waitTableDisabled(byte[] table) throws InterruptedException, IOException { 3197 waitTableDisabled(table, 30000); 3198 } 3199 3200 public void waitTableDisabled(TableName table, long millisTimeout) 3201 throws InterruptedException, IOException { 3202 waitFor(millisTimeout, predicateTableDisabled(table)); 3203 } 3204 3205 /** 3206 * Waits for a table to be 'disabled'. Disabled means that table is set as 'disabled' 3207 * @param table Table to wait on. 3208 * @param timeoutMillis Time to wait on it being marked disabled. 3209 */ 3210 public void waitTableDisabled(byte[] table, long timeoutMillis) 3211 throws InterruptedException, IOException { 3212 waitTableDisabled(TableName.valueOf(table), timeoutMillis); 3213 } 3214 3215 /** 3216 * Make sure that at least the specified number of region servers are running 3217 * @param num minimum number of region servers that should be running 3218 * @return true if we started some servers 3219 */ 3220 public boolean ensureSomeRegionServersAvailable(final int num) throws IOException { 3221 boolean startedServer = false; 3222 MiniHBaseCluster hbaseCluster = getMiniHBaseCluster(); 3223 for (int i = hbaseCluster.getLiveRegionServerThreads().size(); i < num; ++i) { 3224 LOG.info("Started new server=" + hbaseCluster.startRegionServer()); 3225 startedServer = true; 3226 } 3227 3228 return startedServer; 3229 } 3230 3231 /** 3232 * Make sure that at least the specified number of region servers are running. We don't count the 3233 * ones that are currently stopping or are stopped. 3234 * @param num minimum number of region servers that should be running 3235 * @return true if we started some servers 3236 */ 3237 public boolean ensureSomeNonStoppedRegionServersAvailable(final int num) throws IOException { 3238 boolean startedServer = ensureSomeRegionServersAvailable(num); 3239 3240 int nonStoppedServers = 0; 3241 for (JVMClusterUtil.RegionServerThread rst : getMiniHBaseCluster().getRegionServerThreads()) { 3242 3243 HRegionServer hrs = rst.getRegionServer(); 3244 if (hrs.isStopping() || hrs.isStopped()) { 3245 LOG.info("A region server is stopped or stopping:" + hrs); 3246 } else { 3247 nonStoppedServers++; 3248 } 3249 } 3250 for (int i = nonStoppedServers; i < num; ++i) { 3251 LOG.info("Started new server=" + getMiniHBaseCluster().startRegionServer()); 3252 startedServer = true; 3253 } 3254 return startedServer; 3255 } 3256 3257 /** 3258 * This method clones the passed <code>c</code> configuration setting a new user into the clone. 3259 * Use it getting new instances of FileSystem. Only works for DistributedFileSystem w/o Kerberos. 3260 * @param c Initial configuration 3261 * @param differentiatingSuffix Suffix to differentiate this user from others. 3262 * @return A new configuration instance with a different user set into it. 3263 */ 3264 public static User getDifferentUser(final Configuration c, final String differentiatingSuffix) 3265 throws IOException { 3266 FileSystem currentfs = FileSystem.get(c); 3267 if (!(currentfs instanceof DistributedFileSystem) || User.isHBaseSecurityEnabled(c)) { 3268 return User.getCurrent(); 3269 } 3270 // Else distributed filesystem. Make a new instance per daemon. Below 3271 // code is taken from the AppendTestUtil over in hdfs. 3272 String username = User.getCurrent().getName() + differentiatingSuffix; 3273 User user = User.createUserForTesting(c, username, new String[] { "supergroup" }); 3274 return user; 3275 } 3276 3277 public static NavigableSet<String> getAllOnlineRegions(MiniHBaseCluster cluster) 3278 throws IOException { 3279 NavigableSet<String> online = new TreeSet<>(); 3280 for (RegionServerThread rst : cluster.getLiveRegionServerThreads()) { 3281 try { 3282 for (RegionInfo region : ProtobufUtil 3283 .getOnlineRegions(rst.getRegionServer().getRSRpcServices())) { 3284 online.add(region.getRegionNameAsString()); 3285 } 3286 } catch (RegionServerStoppedException e) { 3287 // That's fine. 3288 } 3289 } 3290 return online; 3291 } 3292 3293 /** 3294 * Set maxRecoveryErrorCount in DFSClient. In 0.20 pre-append its hard-coded to 5 and makes tests 3295 * linger. Here is the exception you'll see: 3296 * 3297 * <pre> 3298 * 2010-06-15 11:52:28,511 WARN [DataStreamer for file /hbase/.logs/wal.1276627923013 block 3299 * blk_928005470262850423_1021] hdfs.DFSClient$DFSOutputStream(2657): Error Recovery for block 3300 * blk_928005470262850423_1021 failed because recovery from primary datanode 127.0.0.1:53683 3301 * failed 4 times. Pipeline was 127.0.0.1:53687, 127.0.0.1:53683. Will retry... 3302 * </pre> 3303 * 3304 * @param stream A DFSClient.DFSOutputStream. 3305 */ 3306 public static void setMaxRecoveryErrorCount(final OutputStream stream, final int max) { 3307 try { 3308 Class<?>[] clazzes = DFSClient.class.getDeclaredClasses(); 3309 for (Class<?> clazz : clazzes) { 3310 String className = clazz.getSimpleName(); 3311 if (className.equals("DFSOutputStream")) { 3312 if (clazz.isInstance(stream)) { 3313 Field maxRecoveryErrorCountField = 3314 stream.getClass().getDeclaredField("maxRecoveryErrorCount"); 3315 maxRecoveryErrorCountField.setAccessible(true); 3316 maxRecoveryErrorCountField.setInt(stream, max); 3317 break; 3318 } 3319 } 3320 } 3321 } catch (Exception e) { 3322 LOG.info("Could not set max recovery field", e); 3323 } 3324 } 3325 3326 /** 3327 * Uses directly the assignment manager to assign the region. and waits until the specified region 3328 * has completed assignment. 3329 * @return true if the region is assigned false otherwise. 3330 */ 3331 public boolean assignRegion(final RegionInfo regionInfo) 3332 throws IOException, InterruptedException { 3333 final AssignmentManager am = getHBaseCluster().getMaster().getAssignmentManager(); 3334 am.assign(regionInfo); 3335 return AssignmentTestingUtil.waitForAssignment(am, regionInfo); 3336 } 3337 3338 /** 3339 * Move region to destination server and wait till region is completely moved and online 3340 * @param destRegion region to move 3341 * @param destServer destination server of the region 3342 */ 3343 public void moveRegionAndWait(RegionInfo destRegion, ServerName destServer) 3344 throws InterruptedException, IOException { 3345 HMaster master = getMiniHBaseCluster().getMaster(); 3346 // TODO: Here we start the move. The move can take a while. 3347 getAdmin().move(destRegion.getEncodedNameAsBytes(), destServer); 3348 while (true) { 3349 ServerName serverName = 3350 master.getAssignmentManager().getRegionStates().getRegionServerOfRegion(destRegion); 3351 if (serverName != null && serverName.equals(destServer)) { 3352 assertRegionOnServer(destRegion, serverName, 2000); 3353 break; 3354 } 3355 Thread.sleep(10); 3356 } 3357 } 3358 3359 /** 3360 * Wait until all regions for a table in hbase:meta have a non-empty info:server, up to a 3361 * configuable timeout value (default is 60 seconds) This means all regions have been deployed, 3362 * master has been informed and updated hbase:meta with the regions deployed server. 3363 * @param tableName the table name 3364 */ 3365 public void waitUntilAllRegionsAssigned(final TableName tableName) throws IOException { 3366 waitUntilAllRegionsAssigned(tableName, 3367 this.conf.getLong("hbase.client.sync.wait.timeout.msec", 60000)); 3368 } 3369 3370 /** 3371 * Waith until all system table's regions get assigned 3372 */ 3373 public void waitUntilAllSystemRegionsAssigned() throws IOException { 3374 waitUntilAllRegionsAssigned(TableName.META_TABLE_NAME); 3375 } 3376 3377 /** 3378 * Wait until all regions for a table in hbase:meta have a non-empty info:server, or until 3379 * timeout. This means all regions have been deployed, master has been informed and updated 3380 * hbase:meta with the regions deployed server. 3381 * @param tableName the table name 3382 * @param timeout timeout, in milliseconds 3383 */ 3384 public void waitUntilAllRegionsAssigned(final TableName tableName, final long timeout) 3385 throws IOException { 3386 if (!TableName.isMetaTableName(tableName)) { 3387 try (final Table meta = getConnection().getTable(TableName.META_TABLE_NAME)) { 3388 LOG.debug("Waiting until all regions of table " + tableName + " get assigned. Timeout = " 3389 + timeout + "ms"); 3390 waitFor(timeout, 200, true, new ExplainingPredicate<IOException>() { 3391 @Override 3392 public String explainFailure() throws IOException { 3393 return explainTableAvailability(tableName); 3394 } 3395 3396 @Override 3397 public boolean evaluate() throws IOException { 3398 Scan scan = new Scan(); 3399 scan.addFamily(HConstants.CATALOG_FAMILY); 3400 boolean tableFound = false; 3401 try (ResultScanner s = meta.getScanner(scan)) { 3402 for (Result r; (r = s.next()) != null;) { 3403 byte[] b = r.getValue(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER); 3404 RegionInfo info = RegionInfo.parseFromOrNull(b); 3405 if (info != null && info.getTable().equals(tableName)) { 3406 // Get server hosting this region from catalog family. Return false if no server 3407 // hosting this region, or if the server hosting this region was recently killed 3408 // (for fault tolerance testing). 3409 tableFound = true; 3410 byte[] server = 3411 r.getValue(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER); 3412 if (server == null) { 3413 return false; 3414 } else { 3415 byte[] startCode = 3416 r.getValue(HConstants.CATALOG_FAMILY, HConstants.STARTCODE_QUALIFIER); 3417 ServerName serverName = 3418 ServerName.valueOf(Bytes.toString(server).replaceFirst(":", ",") + "," 3419 + Bytes.toLong(startCode)); 3420 if ( 3421 !getHBaseClusterInterface().isDistributedCluster() 3422 && getHBaseCluster().isKilledRS(serverName) 3423 ) { 3424 return false; 3425 } 3426 } 3427 if (RegionStateStore.getRegionState(r, info) != RegionState.State.OPEN) { 3428 return false; 3429 } 3430 } 3431 } 3432 } 3433 if (!tableFound) { 3434 LOG.warn( 3435 "Didn't find the entries for table " + tableName + " in meta, already deleted?"); 3436 } 3437 return tableFound; 3438 } 3439 }); 3440 } 3441 } 3442 LOG.info("All regions for table " + tableName + " assigned to meta. Checking AM states."); 3443 // check from the master state if we are using a mini cluster 3444 if (!getHBaseClusterInterface().isDistributedCluster()) { 3445 // So, all regions are in the meta table but make sure master knows of the assignments before 3446 // returning -- sometimes this can lag. 3447 HMaster master = getHBaseCluster().getMaster(); 3448 final RegionStates states = master.getAssignmentManager().getRegionStates(); 3449 waitFor(timeout, 200, new ExplainingPredicate<IOException>() { 3450 @Override 3451 public String explainFailure() throws IOException { 3452 return explainTableAvailability(tableName); 3453 } 3454 3455 @Override 3456 public boolean evaluate() throws IOException { 3457 List<RegionInfo> hris = states.getRegionsOfTable(tableName); 3458 return hris != null && !hris.isEmpty(); 3459 } 3460 }); 3461 } 3462 LOG.info("All regions for table " + tableName + " assigned."); 3463 } 3464 3465 /** 3466 * Do a small get/scan against one store. This is required because store has no actual methods of 3467 * querying itself, and relies on StoreScanner. 3468 */ 3469 public static List<Cell> getFromStoreFile(HStore store, Get get) throws IOException { 3470 Scan scan = new Scan(get); 3471 InternalScanner scanner = (InternalScanner) store.getScanner(scan, 3472 scan.getFamilyMap().get(store.getColumnFamilyDescriptor().getName()), 3473 // originally MultiVersionConcurrencyControl.resetThreadReadPoint() was called to set 3474 // readpoint 0. 3475 0); 3476 3477 List<Cell> result = new ArrayList<>(); 3478 scanner.next(result); 3479 if (!result.isEmpty()) { 3480 // verify that we are on the row we want: 3481 Cell kv = result.get(0); 3482 if (!CellUtil.matchingRows(kv, get.getRow())) { 3483 result.clear(); 3484 } 3485 } 3486 scanner.close(); 3487 return result; 3488 } 3489 3490 /** 3491 * Create region split keys between startkey and endKey 3492 * @param numRegions the number of regions to be created. it has to be greater than 3. 3493 * @return resulting split keys 3494 */ 3495 public byte[][] getRegionSplitStartKeys(byte[] startKey, byte[] endKey, int numRegions) { 3496 if (numRegions <= 3) { 3497 throw new AssertionError(); 3498 } 3499 byte[][] tmpSplitKeys = Bytes.split(startKey, endKey, numRegions - 3); 3500 byte[][] result = new byte[tmpSplitKeys.length + 1][]; 3501 System.arraycopy(tmpSplitKeys, 0, result, 1, tmpSplitKeys.length); 3502 result[0] = HConstants.EMPTY_BYTE_ARRAY; 3503 return result; 3504 } 3505 3506 /** 3507 * Do a small get/scan against one store. This is required because store has no actual methods of 3508 * querying itself, and relies on StoreScanner. 3509 */ 3510 public static List<Cell> getFromStoreFile(HStore store, byte[] row, NavigableSet<byte[]> columns) 3511 throws IOException { 3512 Get get = new Get(row); 3513 Map<byte[], NavigableSet<byte[]>> s = get.getFamilyMap(); 3514 s.put(store.getColumnFamilyDescriptor().getName(), columns); 3515 3516 return getFromStoreFile(store, get); 3517 } 3518 3519 public static void assertKVListsEqual(String additionalMsg, final List<? extends Cell> expected, 3520 final List<? extends Cell> actual) { 3521 final int eLen = expected.size(); 3522 final int aLen = actual.size(); 3523 final int minLen = Math.min(eLen, aLen); 3524 3525 int i; 3526 for (i = 0; i < minLen 3527 && CellComparator.getInstance().compare(expected.get(i), actual.get(i)) == 0; ++i) { 3528 } 3529 3530 if (additionalMsg == null) { 3531 additionalMsg = ""; 3532 } 3533 if (!additionalMsg.isEmpty()) { 3534 additionalMsg = ". " + additionalMsg; 3535 } 3536 3537 if (eLen != aLen || i != minLen) { 3538 throw new AssertionError("Expected and actual KV arrays differ at position " + i + ": " 3539 + safeGetAsStr(expected, i) + " (length " + eLen + ") vs. " + safeGetAsStr(actual, i) 3540 + " (length " + aLen + ")" + additionalMsg); 3541 } 3542 } 3543 3544 public static <T> String safeGetAsStr(List<T> lst, int i) { 3545 if (0 <= i && i < lst.size()) { 3546 return lst.get(i).toString(); 3547 } else { 3548 return "<out_of_range>"; 3549 } 3550 } 3551 3552 public String getRpcConnnectionURI() throws UnknownHostException { 3553 return "hbase+rpc://" + MasterRegistry.getMasterAddr(conf); 3554 } 3555 3556 public String getZkConnectionURI() { 3557 return "hbase+zk://" + conf.get(HConstants.ZOOKEEPER_QUORUM) + ":" 3558 + conf.get(HConstants.ZOOKEEPER_CLIENT_PORT) 3559 + conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT, HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT); 3560 } 3561 3562 /** 3563 * Get the zk based cluster key for this cluster. 3564 * @deprecated since 2.7.0, will be removed in 4.0.0. Now we use connection uri to specify the 3565 * connection info of a cluster. Keep here only for compatibility. 3566 * @see #getRpcConnnectionURI() 3567 * @see #getZkConnectionURI() 3568 */ 3569 @Deprecated 3570 public String getClusterKey() { 3571 return conf.get(HConstants.ZOOKEEPER_QUORUM) + ":" + conf.get(HConstants.ZOOKEEPER_CLIENT_PORT) 3572 + ":" 3573 + conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT, HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT); 3574 } 3575 3576 /** Creates a random table with the given parameters */ 3577 public Table createRandomTable(TableName tableName, final Collection<String> families, 3578 final int maxVersions, final int numColsPerRow, final int numFlushes, final int numRegions, 3579 final int numRowsPerFlush) throws IOException, InterruptedException { 3580 3581 LOG.info("\n\nCreating random table " + tableName + " with " + numRegions + " regions, " 3582 + numFlushes + " storefiles per region, " + numRowsPerFlush + " rows per flush, maxVersions=" 3583 + maxVersions + "\n"); 3584 3585 final int numCF = families.size(); 3586 final byte[][] cfBytes = new byte[numCF][]; 3587 { 3588 int cfIndex = 0; 3589 for (String cf : families) { 3590 cfBytes[cfIndex++] = Bytes.toBytes(cf); 3591 } 3592 } 3593 3594 final int actualStartKey = 0; 3595 final int actualEndKey = Integer.MAX_VALUE; 3596 final int keysPerRegion = (actualEndKey - actualStartKey) / numRegions; 3597 final int splitStartKey = actualStartKey + keysPerRegion; 3598 final int splitEndKey = actualEndKey - keysPerRegion; 3599 final String keyFormat = "%08x"; 3600 final Table table = createTable(tableName, cfBytes, maxVersions, 3601 Bytes.toBytes(String.format(keyFormat, splitStartKey)), 3602 Bytes.toBytes(String.format(keyFormat, splitEndKey)), numRegions); 3603 3604 if (hbaseCluster != null) { 3605 getMiniHBaseCluster().flushcache(TableName.META_TABLE_NAME); 3606 } 3607 3608 BufferedMutator mutator = getConnection().getBufferedMutator(tableName); 3609 3610 final Random rand = ThreadLocalRandom.current(); 3611 for (int iFlush = 0; iFlush < numFlushes; ++iFlush) { 3612 for (int iRow = 0; iRow < numRowsPerFlush; ++iRow) { 3613 final byte[] row = Bytes.toBytes( 3614 String.format(keyFormat, actualStartKey + rand.nextInt(actualEndKey - actualStartKey))); 3615 3616 Put put = new Put(row); 3617 Delete del = new Delete(row); 3618 for (int iCol = 0; iCol < numColsPerRow; ++iCol) { 3619 final byte[] cf = cfBytes[rand.nextInt(numCF)]; 3620 final long ts = rand.nextInt(); 3621 final byte[] qual = Bytes.toBytes("col" + iCol); 3622 if (rand.nextBoolean()) { 3623 final byte[] value = 3624 Bytes.toBytes("value_for_row_" + iRow + "_cf_" + Bytes.toStringBinary(cf) + "_col_" 3625 + iCol + "_ts_" + ts + "_random_" + rand.nextLong()); 3626 put.addColumn(cf, qual, ts, value); 3627 } else if (rand.nextDouble() < 0.8) { 3628 del.addColumn(cf, qual, ts); 3629 } else { 3630 del.addColumns(cf, qual, ts); 3631 } 3632 } 3633 3634 if (!put.isEmpty()) { 3635 mutator.mutate(put); 3636 } 3637 3638 if (!del.isEmpty()) { 3639 mutator.mutate(del); 3640 } 3641 } 3642 LOG.info("Initiating flush #" + iFlush + " for table " + tableName); 3643 mutator.flush(); 3644 if (hbaseCluster != null) { 3645 getMiniHBaseCluster().flushcache(table.getName()); 3646 } 3647 } 3648 mutator.close(); 3649 3650 return table; 3651 } 3652 3653 public static int randomFreePort() { 3654 return HBaseCommonTestingUtility.randomFreePort(); 3655 } 3656 3657 public static String randomMultiCastAddress() { 3658 return "226.1.1." + ThreadLocalRandom.current().nextInt(254); 3659 } 3660 3661 public static void waitForHostPort(String host, int port) throws IOException { 3662 final int maxTimeMs = 10000; 3663 final int maxNumAttempts = maxTimeMs / HConstants.SOCKET_RETRY_WAIT_MS; 3664 IOException savedException = null; 3665 LOG.info("Waiting for server at " + host + ":" + port); 3666 for (int attempt = 0; attempt < maxNumAttempts; ++attempt) { 3667 try { 3668 Socket sock = new Socket(InetAddress.getByName(host), port); 3669 sock.close(); 3670 savedException = null; 3671 LOG.info("Server at " + host + ":" + port + " is available"); 3672 break; 3673 } catch (UnknownHostException e) { 3674 throw new IOException("Failed to look up " + host, e); 3675 } catch (IOException e) { 3676 savedException = e; 3677 } 3678 Threads.sleepWithoutInterrupt(HConstants.SOCKET_RETRY_WAIT_MS); 3679 } 3680 3681 if (savedException != null) { 3682 throw savedException; 3683 } 3684 } 3685 3686 /** 3687 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3688 * continues. 3689 * @return the number of regions the table was split into 3690 */ 3691 public static int createPreSplitLoadTestTable(Configuration conf, TableName tableName, 3692 byte[] columnFamily, Algorithm compression, DataBlockEncoding dataBlockEncoding) 3693 throws IOException { 3694 return createPreSplitLoadTestTable(conf, tableName, columnFamily, compression, 3695 dataBlockEncoding, DEFAULT_REGIONS_PER_SERVER, 1, Durability.USE_DEFAULT); 3696 } 3697 3698 /** 3699 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3700 * continues. 3701 * @return the number of regions the table was split into 3702 */ 3703 public static int createPreSplitLoadTestTable(Configuration conf, TableName tableName, 3704 byte[] columnFamily, Algorithm compression, DataBlockEncoding dataBlockEncoding, 3705 int numRegionsPerServer, int regionReplication, Durability durability) throws IOException { 3706 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 3707 builder.setDurability(durability); 3708 builder.setRegionReplication(regionReplication); 3709 ColumnFamilyDescriptorBuilder cfBuilder = 3710 ColumnFamilyDescriptorBuilder.newBuilder(columnFamily); 3711 cfBuilder.setDataBlockEncoding(dataBlockEncoding); 3712 cfBuilder.setCompressionType(compression); 3713 return createPreSplitLoadTestTable(conf, builder.build(), cfBuilder.build(), 3714 numRegionsPerServer); 3715 } 3716 3717 /** 3718 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3719 * continues. 3720 * @return the number of regions the table was split into 3721 */ 3722 public static int createPreSplitLoadTestTable(Configuration conf, TableName tableName, 3723 byte[][] columnFamilies, Algorithm compression, DataBlockEncoding dataBlockEncoding, 3724 int numRegionsPerServer, int regionReplication, Durability durability) throws IOException { 3725 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); 3726 builder.setDurability(durability); 3727 builder.setRegionReplication(regionReplication); 3728 ColumnFamilyDescriptor[] hcds = new ColumnFamilyDescriptor[columnFamilies.length]; 3729 for (int i = 0; i < columnFamilies.length; i++) { 3730 ColumnFamilyDescriptorBuilder cfBuilder = 3731 ColumnFamilyDescriptorBuilder.newBuilder(columnFamilies[i]); 3732 cfBuilder.setDataBlockEncoding(dataBlockEncoding); 3733 cfBuilder.setCompressionType(compression); 3734 hcds[i] = cfBuilder.build(); 3735 } 3736 return createPreSplitLoadTestTable(conf, builder.build(), hcds, numRegionsPerServer); 3737 } 3738 3739 /** 3740 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3741 * continues. 3742 * @return the number of regions the table was split into 3743 */ 3744 public static int createPreSplitLoadTestTable(Configuration conf, TableDescriptor desc, 3745 ColumnFamilyDescriptor hcd) throws IOException { 3746 return createPreSplitLoadTestTable(conf, desc, hcd, DEFAULT_REGIONS_PER_SERVER); 3747 } 3748 3749 /** 3750 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3751 * continues. 3752 * @return the number of regions the table was split into 3753 */ 3754 public static int createPreSplitLoadTestTable(Configuration conf, TableDescriptor desc, 3755 ColumnFamilyDescriptor hcd, int numRegionsPerServer) throws IOException { 3756 return createPreSplitLoadTestTable(conf, desc, new ColumnFamilyDescriptor[] { hcd }, 3757 numRegionsPerServer); 3758 } 3759 3760 /** 3761 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3762 * continues. 3763 * @return the number of regions the table was split into 3764 */ 3765 public static int createPreSplitLoadTestTable(Configuration conf, TableDescriptor desc, 3766 ColumnFamilyDescriptor[] hcds, int numRegionsPerServer) throws IOException { 3767 return createPreSplitLoadTestTable(conf, desc, hcds, new RegionSplitter.HexStringSplit(), 3768 numRegionsPerServer); 3769 } 3770 3771 /** 3772 * Creates a pre-split table for load testing. If the table already exists, logs a warning and 3773 * continues. 3774 * @return the number of regions the table was split into 3775 */ 3776 public static int createPreSplitLoadTestTable(Configuration conf, TableDescriptor td, 3777 ColumnFamilyDescriptor[] cds, SplitAlgorithm splitter, int numRegionsPerServer) 3778 throws IOException { 3779 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(td); 3780 for (ColumnFamilyDescriptor cd : cds) { 3781 if (!td.hasColumnFamily(cd.getName())) { 3782 builder.setColumnFamily(cd); 3783 } 3784 } 3785 td = builder.build(); 3786 int totalNumberOfRegions = 0; 3787 Connection unmanagedConnection = ConnectionFactory.createConnection(conf); 3788 Admin admin = unmanagedConnection.getAdmin(); 3789 3790 try { 3791 // create a table a pre-splits regions. 3792 // The number of splits is set as: 3793 // region servers * regions per region server). 3794 int numberOfServers = admin.getRegionServers().size(); 3795 if (numberOfServers == 0) { 3796 throw new IllegalStateException("No live regionservers"); 3797 } 3798 3799 totalNumberOfRegions = numberOfServers * numRegionsPerServer; 3800 LOG.info("Number of live regionservers: " + numberOfServers + ", " 3801 + "pre-splitting table into " + totalNumberOfRegions + " regions " + "(regions per server: " 3802 + numRegionsPerServer + ")"); 3803 3804 byte[][] splits = splitter.split(totalNumberOfRegions); 3805 3806 admin.createTable(td, splits); 3807 } catch (MasterNotRunningException e) { 3808 LOG.error("Master not running", e); 3809 throw new IOException(e); 3810 } catch (TableExistsException e) { 3811 LOG.warn("Table " + td.getTableName() + " already exists, continuing"); 3812 } finally { 3813 admin.close(); 3814 unmanagedConnection.close(); 3815 } 3816 return totalNumberOfRegions; 3817 } 3818 3819 public static int getMetaRSPort(Connection connection) throws IOException { 3820 try (RegionLocator locator = connection.getRegionLocator(TableName.META_TABLE_NAME)) { 3821 return locator.getRegionLocation(Bytes.toBytes("")).getPort(); 3822 } 3823 } 3824 3825 /** 3826 * Due to async racing issue, a region may not be in the online region list of a region server 3827 * yet, after the assignment znode is deleted and the new assignment is recorded in master. 3828 */ 3829 public void assertRegionOnServer(final RegionInfo hri, final ServerName server, 3830 final long timeout) throws IOException, InterruptedException { 3831 long timeoutTime = EnvironmentEdgeManager.currentTime() + timeout; 3832 while (true) { 3833 List<RegionInfo> regions = getAdmin().getRegions(server); 3834 if (regions.stream().anyMatch(r -> RegionInfo.COMPARATOR.compare(r, hri) == 0)) return; 3835 long now = EnvironmentEdgeManager.currentTime(); 3836 if (now > timeoutTime) break; 3837 Thread.sleep(10); 3838 } 3839 throw new AssertionError( 3840 "Could not find region " + hri.getRegionNameAsString() + " on server " + server); 3841 } 3842 3843 /** 3844 * Check to make sure the region is open on the specified region server, but not on any other one. 3845 */ 3846 public void assertRegionOnlyOnServer(final RegionInfo hri, final ServerName server, 3847 final long timeout) throws IOException, InterruptedException { 3848 long timeoutTime = EnvironmentEdgeManager.currentTime() + timeout; 3849 while (true) { 3850 List<RegionInfo> regions = getAdmin().getRegions(server); 3851 if (regions.stream().anyMatch(r -> RegionInfo.COMPARATOR.compare(r, hri) == 0)) { 3852 List<JVMClusterUtil.RegionServerThread> rsThreads = 3853 getHBaseCluster().getLiveRegionServerThreads(); 3854 for (JVMClusterUtil.RegionServerThread rsThread : rsThreads) { 3855 HRegionServer rs = rsThread.getRegionServer(); 3856 if (server.equals(rs.getServerName())) { 3857 continue; 3858 } 3859 Collection<HRegion> hrs = rs.getOnlineRegionsLocalContext(); 3860 for (HRegion r : hrs) { 3861 if (r.getRegionInfo().getRegionId() == hri.getRegionId()) { 3862 throw new AssertionError("Region should not be double assigned"); 3863 } 3864 } 3865 } 3866 return; // good, we are happy 3867 } 3868 long now = EnvironmentEdgeManager.currentTime(); 3869 if (now > timeoutTime) break; 3870 Thread.sleep(10); 3871 } 3872 throw new AssertionError( 3873 "Could not find region " + hri.getRegionNameAsString() + " on server " + server); 3874 } 3875 3876 public HRegion createTestRegion(String tableName, ColumnFamilyDescriptor cd) throws IOException { 3877 TableDescriptor td = 3878 TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName)).setColumnFamily(cd).build(); 3879 RegionInfo info = RegionInfoBuilder.newBuilder(TableName.valueOf(tableName)).build(); 3880 return createRegionAndWAL(info, getDataTestDir(), getConfiguration(), td); 3881 } 3882 3883 public HRegion createTestRegion(String tableName, ColumnFamilyDescriptor cd, 3884 BlockCache blockCache) throws IOException { 3885 TableDescriptor td = 3886 TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName)).setColumnFamily(cd).build(); 3887 RegionInfo info = RegionInfoBuilder.newBuilder(TableName.valueOf(tableName)).build(); 3888 return createRegionAndWAL(info, getDataTestDir(), getConfiguration(), td, blockCache); 3889 } 3890 3891 public static void setFileSystemURI(String fsURI) { 3892 FS_URI = fsURI; 3893 } 3894 3895 /** 3896 * Returns a {@link Predicate} for checking that there are no regions in transition in master 3897 */ 3898 public ExplainingPredicate<IOException> predicateNoRegionsInTransition() { 3899 return new ExplainingPredicate<IOException>() { 3900 @Override 3901 public String explainFailure() throws IOException { 3902 final AssignmentManager am = getMiniHBaseCluster().getMaster().getAssignmentManager(); 3903 return "found in transition: " + am.getRegionsInTransition().toString(); 3904 } 3905 3906 @Override 3907 public boolean evaluate() throws IOException { 3908 HMaster master = getMiniHBaseCluster().getMaster(); 3909 if (master == null) return false; 3910 AssignmentManager am = master.getAssignmentManager(); 3911 if (am == null) return false; 3912 return !am.hasRegionsInTransition(); 3913 } 3914 }; 3915 } 3916 3917 /** 3918 * Returns a {@link Predicate} for checking that table is enabled 3919 */ 3920 public Waiter.Predicate<IOException> predicateTableEnabled(final TableName tableName) { 3921 return new ExplainingPredicate<IOException>() { 3922 @Override 3923 public String explainFailure() throws IOException { 3924 return explainTableState(tableName, TableState.State.ENABLED); 3925 } 3926 3927 @Override 3928 public boolean evaluate() throws IOException { 3929 return getAdmin().tableExists(tableName) && getAdmin().isTableEnabled(tableName); 3930 } 3931 }; 3932 } 3933 3934 /** 3935 * Returns a {@link Predicate} for checking that table is enabled 3936 */ 3937 public Waiter.Predicate<IOException> predicateTableDisabled(final TableName tableName) { 3938 return new ExplainingPredicate<IOException>() { 3939 @Override 3940 public String explainFailure() throws IOException { 3941 return explainTableState(tableName, TableState.State.DISABLED); 3942 } 3943 3944 @Override 3945 public boolean evaluate() throws IOException { 3946 return getAdmin().isTableDisabled(tableName); 3947 } 3948 }; 3949 } 3950 3951 /** 3952 * Returns a {@link Predicate} for checking that table is enabled 3953 */ 3954 public Waiter.Predicate<IOException> predicateTableAvailable(final TableName tableName) { 3955 return new ExplainingPredicate<IOException>() { 3956 @Override 3957 public String explainFailure() throws IOException { 3958 return explainTableAvailability(tableName); 3959 } 3960 3961 @Override 3962 public boolean evaluate() throws IOException { 3963 boolean tableAvailable = getAdmin().isTableAvailable(tableName); 3964 if (tableAvailable) { 3965 try (Table table = getConnection().getTable(tableName)) { 3966 TableDescriptor htd = table.getDescriptor(); 3967 for (HRegionLocation loc : getConnection().getRegionLocator(tableName) 3968 .getAllRegionLocations()) { 3969 Scan scan = new Scan().withStartRow(loc.getRegion().getStartKey()) 3970 .withStopRow(loc.getRegion().getEndKey()).setOneRowLimit() 3971 .setMaxResultsPerColumnFamily(1).setCacheBlocks(false); 3972 for (byte[] family : htd.getColumnFamilyNames()) { 3973 scan.addFamily(family); 3974 } 3975 try (ResultScanner scanner = table.getScanner(scan)) { 3976 scanner.next(); 3977 } 3978 } 3979 } 3980 } 3981 return tableAvailable; 3982 } 3983 }; 3984 } 3985 3986 /** 3987 * Wait until no regions in transition. 3988 * @param timeout How long to wait. 3989 */ 3990 public void waitUntilNoRegionsInTransition(final long timeout) throws IOException { 3991 waitFor(timeout, predicateNoRegionsInTransition()); 3992 } 3993 3994 /** 3995 * Wait until no regions in transition. (time limit 15min) 3996 */ 3997 public void waitUntilNoRegionsInTransition() throws IOException { 3998 waitUntilNoRegionsInTransition(15 * 60000); 3999 } 4000 4001 /** 4002 * Wait until labels is ready in VisibilityLabelsCache. 4003 */ 4004 public void waitLabelAvailable(long timeoutMillis, final String... labels) { 4005 final VisibilityLabelsCache labelsCache = VisibilityLabelsCache.get(); 4006 waitFor(timeoutMillis, new Waiter.ExplainingPredicate<RuntimeException>() { 4007 4008 @Override 4009 public boolean evaluate() { 4010 for (String label : labels) { 4011 if (labelsCache.getLabelOrdinal(label) == 0) { 4012 return false; 4013 } 4014 } 4015 return true; 4016 } 4017 4018 @Override 4019 public String explainFailure() { 4020 for (String label : labels) { 4021 if (labelsCache.getLabelOrdinal(label) == 0) { 4022 return label + " is not available yet"; 4023 } 4024 } 4025 return ""; 4026 } 4027 }); 4028 } 4029 4030 /** 4031 * Create a set of column descriptors with the combination of compression, encoding, bloom codecs 4032 * available. 4033 * @return the list of column descriptors 4034 */ 4035 public static List<ColumnFamilyDescriptor> generateColumnDescriptors() { 4036 return generateColumnDescriptors(""); 4037 } 4038 4039 /** 4040 * Create a set of column descriptors with the combination of compression, encoding, bloom codecs 4041 * available. 4042 * @param prefix family names prefix 4043 * @return the list of column descriptors 4044 */ 4045 public static List<ColumnFamilyDescriptor> generateColumnDescriptors(final String prefix) { 4046 List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>(); 4047 long familyId = 0; 4048 for (Compression.Algorithm compressionType : getSupportedCompressionAlgorithms()) { 4049 for (DataBlockEncoding encodingType : DataBlockEncoding.values()) { 4050 for (BloomType bloomType : BloomType.values()) { 4051 String name = String.format("%s-cf-!@#&-%d!@#", prefix, familyId); 4052 ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = 4053 ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(name)); 4054 columnFamilyDescriptorBuilder.setCompressionType(compressionType); 4055 columnFamilyDescriptorBuilder.setDataBlockEncoding(encodingType); 4056 columnFamilyDescriptorBuilder.setBloomFilterType(bloomType); 4057 columnFamilyDescriptors.add(columnFamilyDescriptorBuilder.build()); 4058 familyId++; 4059 } 4060 } 4061 } 4062 return columnFamilyDescriptors; 4063 } 4064 4065 /** 4066 * Get supported compression algorithms. 4067 * @return supported compression algorithms. 4068 */ 4069 public static Compression.Algorithm[] getSupportedCompressionAlgorithms() { 4070 String[] allAlgos = HFile.getSupportedCompressionAlgorithms(); 4071 List<Compression.Algorithm> supportedAlgos = new ArrayList<>(); 4072 for (String algoName : allAlgos) { 4073 try { 4074 Compression.Algorithm algo = Compression.getCompressionAlgorithmByName(algoName); 4075 algo.getCompressor(); 4076 supportedAlgos.add(algo); 4077 } catch (Throwable t) { 4078 // this algo is not available 4079 } 4080 } 4081 return supportedAlgos.toArray(new Algorithm[supportedAlgos.size()]); 4082 } 4083 4084 public Result getClosestRowBefore(Region r, byte[] row, byte[] family) throws IOException { 4085 Scan scan = new Scan().withStartRow(row); 4086 scan.setReadType(ReadType.PREAD); 4087 scan.setCaching(1); 4088 scan.setReversed(true); 4089 scan.addFamily(family); 4090 try (RegionScanner scanner = r.getScanner(scan)) { 4091 List<Cell> cells = new ArrayList<>(1); 4092 scanner.next(cells); 4093 if (r.getRegionInfo().isMetaRegion() && !isTargetTable(row, cells.get(0))) { 4094 return null; 4095 } 4096 return Result.create(cells); 4097 } 4098 } 4099 4100 private boolean isTargetTable(final byte[] inRow, Cell c) { 4101 String inputRowString = Bytes.toString(inRow); 4102 int i = inputRowString.indexOf(HConstants.DELIMITER); 4103 String outputRowString = Bytes.toString(c.getRowArray(), c.getRowOffset(), c.getRowLength()); 4104 int o = outputRowString.indexOf(HConstants.DELIMITER); 4105 return inputRowString.substring(0, i).equals(outputRowString.substring(0, o)); 4106 } 4107 4108 /** 4109 * Sets up {@link MiniKdc} for testing security. Uses {@link HBaseKerberosUtils} to set the given 4110 * keytab file as {@link HBaseKerberosUtils#KRB_KEYTAB_FILE}. FYI, there is also the easier-to-use 4111 * kerby KDC server and utility for using it, 4112 * {@link org.apache.hadoop.hbase.util.SimpleKdcServerUtil}. The kerby KDC server is preferred; 4113 * less baggage. It came in in HBASE-5291. 4114 */ 4115 public MiniKdc setupMiniKdc(File keytabFile) throws Exception { 4116 Properties conf = MiniKdc.createConf(); 4117 conf.put(MiniKdc.DEBUG, true); 4118 MiniKdc kdc = null; 4119 File dir = null; 4120 // There is time lag between selecting a port and trying to bind with it. It's possible that 4121 // another service captures the port in between which'll result in BindException. 4122 boolean bindException; 4123 int numTries = 0; 4124 do { 4125 try { 4126 bindException = false; 4127 dir = new File(getDataTestDir("kdc").toUri().getPath()); 4128 kdc = new MiniKdc(conf, dir); 4129 kdc.start(); 4130 } catch (BindException e) { 4131 FileUtils.deleteDirectory(dir); // clean directory 4132 numTries++; 4133 if (numTries == 3) { 4134 LOG.error("Failed setting up MiniKDC. Tried " + numTries + " times."); 4135 throw e; 4136 } 4137 LOG.error("BindException encountered when setting up MiniKdc. Trying again."); 4138 bindException = true; 4139 } 4140 } while (bindException); 4141 HBaseKerberosUtils.setKeytabFileForTesting(keytabFile.getAbsolutePath()); 4142 return kdc; 4143 } 4144 4145 public int getNumHFiles(final TableName tableName, final byte[] family) { 4146 int numHFiles = 0; 4147 for (RegionServerThread regionServerThread : getMiniHBaseCluster().getRegionServerThreads()) { 4148 numHFiles += getNumHFilesForRS(regionServerThread.getRegionServer(), tableName, family); 4149 } 4150 return numHFiles; 4151 } 4152 4153 public int getNumHFilesForRS(final HRegionServer rs, final TableName tableName, 4154 final byte[] family) { 4155 int numHFiles = 0; 4156 for (Region region : rs.getRegions(tableName)) { 4157 numHFiles += region.getStore(family).getStorefilesCount(); 4158 } 4159 return numHFiles; 4160 } 4161 4162 private void assertEquals(String message, int expected, int actual) { 4163 if (expected == actual) { 4164 return; 4165 } 4166 String formatted = ""; 4167 if (message != null && !"".equals(message)) { 4168 formatted = message + " "; 4169 } 4170 throw new AssertionError(formatted + "expected:<" + expected + "> but was:<" + actual + ">"); 4171 } 4172 4173 public void verifyTableDescriptorIgnoreTableName(TableDescriptor ltd, TableDescriptor rtd) { 4174 if (ltd.getValues().hashCode() != rtd.getValues().hashCode()) { 4175 throw new AssertionError(); 4176 } 4177 assertEquals("", ltd.getValues().hashCode(), rtd.getValues().hashCode()); 4178 Collection<ColumnFamilyDescriptor> ltdFamilies = Arrays.asList(ltd.getColumnFamilies()); 4179 Collection<ColumnFamilyDescriptor> rtdFamilies = Arrays.asList(rtd.getColumnFamilies()); 4180 assertEquals("", ltdFamilies.size(), rtdFamilies.size()); 4181 for (Iterator<ColumnFamilyDescriptor> it = ltdFamilies.iterator(), 4182 it2 = rtdFamilies.iterator(); it.hasNext();) { 4183 assertEquals("", 0, ColumnFamilyDescriptor.COMPARATOR.compare(it.next(), it2.next())); 4184 } 4185 } 4186 4187 /** 4188 * Await the successful return of {@code condition}, sleeping {@code sleepMillis} between 4189 * invocations. 4190 */ 4191 public static void await(final long sleepMillis, final BooleanSupplier condition) 4192 throws InterruptedException { 4193 try { 4194 while (!condition.getAsBoolean()) { 4195 Thread.sleep(sleepMillis); 4196 } 4197 } catch (RuntimeException e) { 4198 if (e.getCause() instanceof AssertionError) { 4199 throw (AssertionError) e.getCause(); 4200 } 4201 throw e; 4202 } 4203 } 4204}