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