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