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