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