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