org.apache.hadoop.hbase.HConstants#HREGION_OLDLOGDIR_NAME源码实例Demo

@Before
public void setUp() throws Exception {
  this.conf = HBaseConfiguration.create(UTIL.getConfiguration());
  this.conf.setBoolean(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS, false);
  this.fs = UTIL.getDFSCluster().getFileSystem();
  this.rootDir = CommonFSUtils.getRootDir(this.conf);
  this.oldLogDir = new Path(this.rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  String serverName =
      ServerName.valueOf(TEST_NAME.getMethodName() + "-manual", 16010, System.currentTimeMillis())
          .toString();
  this.logName = AbstractFSWALProvider.getWALDirectoryName(serverName);
  this.logDir = new Path(this.rootDir, logName);
  if (UTIL.getDFSCluster().getFileSystem().exists(this.rootDir)) {
    UTIL.getDFSCluster().getFileSystem().delete(this.rootDir, true);
  }
  this.wals = new WALFactory(conf, TEST_NAME.getMethodName());
}
 

@Override
protected void postSetUp() throws IOException {
  Configuration conf = htu.getConfiguration();
  conf.setLong(TimeToLiveMasterLocalStoreWALCleaner.TTL_CONF_KEY, TTL_MS);
  Path testDir = htu.getDataTestDir();
  FileSystem fs = testDir.getFileSystem(conf);
  globalWALArchiveDir = new Path(testDir, HConstants.HREGION_OLDLOGDIR_NAME);
  logCleaner = new LogCleaner(1000, new Stoppable() {

    private volatile boolean stopped = false;

    @Override
    public void stop(String why) {
      stopped = true;
    }

    @Override
    public boolean isStopped() {
      return stopped;
    }
  }, conf, fs, globalWALArchiveDir, cleanerPool);
  choreService.scheduleChore(logCleaner);
}
 

@Before
public void setUp() throws Exception {
  this.conf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
  // this.cluster = TEST_UTIL.getDFSCluster();
  this.fs = TEST_UTIL.getDFSCluster().getFileSystem();
  this.hbaseRootDir = CommonFSUtils.getRootDir(conf);
  this.hbaseWALRootDir = CommonFSUtils.getWALRootDir(conf);
  this.oldLogDir = new Path(this.hbaseWALRootDir,
      HConstants.HREGION_OLDLOGDIR_NAME);
  String serverName = ServerName.valueOf(currentTest.getMethodName(), 16010,
      System.currentTimeMillis()).toString();
  this.logDir = new Path(this.hbaseWALRootDir,
    AbstractFSWALProvider.getWALDirectoryName(serverName));

  if (TEST_UTIL.getDFSCluster().getFileSystem().exists(this.hbaseRootDir)) {
    TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseRootDir, true);
  }
  if (TEST_UTIL.getDFSCluster().getFileSystem().exists(this.hbaseWALRootDir)) {
    TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseWALRootDir, true);
  }
  this.wals = new WALFactory(conf, serverName);
}
 

private int doSearch(Configuration conf, String keysDir) throws Exception {
  Path inputDir = new Path(keysDir);

  getConf().set(SEARCHER_INPUTDIR_KEY, inputDir.toString());
  SortedSet<byte []> keys = readKeysToSearch(getConf());
  if (keys.isEmpty()) throw new RuntimeException("No keys to find");
  LOG.info("Count of keys to find: " + keys.size());
  for(byte [] key: keys)  LOG.info("Key: " + Bytes.toStringBinary(key));
  // Now read all WALs. In two dirs. Presumes certain layout.
  Path walsDir = new Path(CommonFSUtils.getWALRootDir(getConf()), HConstants.HREGION_LOGDIR_NAME);
  Path oldWalsDir = new Path(
      CommonFSUtils.getWALRootDir(getConf()), HConstants.HREGION_OLDLOGDIR_NAME);
  LOG.info("Running Search with keys inputDir=" + inputDir +
    " against " + getConf().get(HConstants.HBASE_DIR));
  int ret = ToolRunner.run(new WALSearcher(getConf()), new String [] {walsDir.toString(), ""});
  if (ret != 0) return ret;
  return ToolRunner.run(new WALSearcher(getConf()), new String [] {oldWalsDir.toString(), ""});
}
 

@Before
public void setUp() throws Exception {
  this.conf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
  this.fs = TEST_UTIL.getDFSCluster().getFileSystem();
  this.hbaseRootDir = CommonFSUtils.getRootDir(this.conf);
  this.oldLogDir = new Path(this.hbaseRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  String serverName =
    ServerName.valueOf(currentTest.getMethodName() + "-manual", 16010, System.currentTimeMillis())
        .toString();
  this.logName = AbstractFSWALProvider.getWALDirectoryName(serverName);
  this.logDir = new Path(this.hbaseRootDir, logName);
  if (TEST_UTIL.getDFSCluster().getFileSystem().exists(this.hbaseRootDir)) {
    TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseRootDir, true);
  }
  this.wals = new WALFactory(conf, currentTest.getMethodName());
}
 

@Before
public void setUp() throws Exception {
  setupCluster();
  Path hbaseRootDir = UTIL.getDataTestDir();
  this.conf = HBaseConfiguration.create(UTIL.getConfiguration());
  this.fs = UTIL.getDFSCluster().getFileSystem();
  this.hbaseRootDir = new Path(this.conf.get(HConstants.HBASE_DIR));
  this.oldLogDir = new Path(this.hbaseRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  this.logDir = new Path(this.hbaseRootDir, HConstants.HREGION_LOGDIR_NAME);
}
 

/**
 * Construct the directory name for all old WALs on a given server. The default old WALs dir looks
 * like: <code>hbase/oldWALs</code>. If you config hbase.separate.oldlogdir.by.regionserver to
 * true, it looks like <code>hbase//oldWALs/kalashnikov.att.net,61634,1486865297088</code>.
 * @param conf
 * @param serverName Server name formatted as described in {@link ServerName}
 * @return the relative WAL directory name
 */
public static String getWALArchiveDirectoryName(Configuration conf, final String serverName) {
  StringBuilder dirName = new StringBuilder(HConstants.HREGION_OLDLOGDIR_NAME);
  if (conf.getBoolean(SEPARATE_OLDLOGDIR, DEFAULT_SEPARATE_OLDLOGDIR)) {
    dirName.append(Path.SEPARATOR);
    dirName.append(serverName);
  }
  return dirName.toString();
}
 

public MasterWalManager(Configuration conf, FileSystem fs, Path rootDir, MasterServices services)
    throws IOException {
  this.fs = fs;
  this.conf = conf;
  this.rootDir = rootDir;
  this.services = services;
  this.splitLogManager = new SplitLogManager(services, conf);

  this.oldLogDir = new Path(rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
}
 

/**
 * Setup WAL log and replication if enabled. Replication setup is done in here because it wants to
 * be hooked up to WAL.
 */
private void setupWALAndReplication() throws IOException {
  boolean isMasterNoTableOrSystemTableOnly = this instanceof HMaster &&
      !LoadBalancer.isMasterCanHostUserRegions(conf);
  WALFactory factory =
      new WALFactory(conf, serverName.toString(), !isMasterNoTableOrSystemTableOnly);
  if (!isMasterNoTableOrSystemTableOnly) {
    // TODO Replication make assumptions here based on the default filesystem impl
    Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
    String logName = AbstractFSWALProvider.getWALDirectoryName(this.serverName.toString());

    Path logDir = new Path(walRootDir, logName);
    LOG.debug("logDir={}", logDir);
    if (this.walFs.exists(logDir)) {
      throw new RegionServerRunningException(
          "Region server has already created directory at " + this.serverName.toString());
    }
    // Always create wal directory as now we need this when master restarts to find out the live
    // region servers.
    if (!this.walFs.mkdirs(logDir)) {
      throw new IOException("Can not create wal directory " + logDir);
    }
    // Instantiate replication if replication enabled. Pass it the log directories.
    createNewReplicationInstance(conf, this, this.walFs, logDir, oldLogDir,
      factory.getWALProvider());
  }
  this.walFactory = factory;
}
 

@BeforeClass
public static void setUpBeforeClass() throws Exception {
  TEST_UTIL.startMiniDFSCluster(1);
  FS = TEST_UTIL.getDFSCluster().getFileSystem();
  Path rootDir = TEST_UTIL.createRootDir();
  oldLogDir = new Path(rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  if (FS.exists(oldLogDir)) FS.delete(oldLogDir, true);
  logDir = new Path(rootDir, HConstants.HREGION_LOGDIR_NAME);
  if (FS.exists(logDir)) FS.delete(logDir, true);
}
 

@Before
public void setUp() throws Exception {
  setupCluster();
  this.conf = HBaseConfiguration.create(UTIL.getConfiguration());
  this.conf.setBoolean(QueryServices.INDEX_FAILURE_THROW_EXCEPTION_ATTRIB, false);
  this.fs = UTIL.getDFSCluster().getFileSystem();
  this.hbaseRootDir = new Path(this.conf.get(HConstants.HBASE_DIR));
  this.oldLogDir = new Path(this.hbaseRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  this.logDir = new Path(this.hbaseRootDir, HConstants.HREGION_LOGDIR_NAME);
}
 

@Before
public void setUp() throws Exception {
  setupCluster();
  this.conf = HBaseConfiguration.create(UTIL.getConfiguration());
  this.fs = UTIL.getDFSCluster().getFileSystem();
  this.hbaseRootDir = new Path(this.conf.get(HConstants.HBASE_DIR));
  this.oldLogDir = new Path(this.hbaseRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  this.logDir = new Path(this.hbaseRootDir, HConstants.HREGION_LOGDIR_NAME);
  // reset the log reader to ensure we pull the one from this config
  HLog.resetLogReaderClass();
}
 

@Test
public void testDifferentRootDirAndWALRootDir() throws Exception {
  // Change wal root dir and reset the configuration
  Path walRootDir = UTIL.createWALRootDir();
  this.conf = HBaseConfiguration.create(UTIL.getConfiguration());

  FileSystem walFs = CommonFSUtils.getWALFileSystem(this.conf);
  this.oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  String serverName =
      ServerName.valueOf(TEST_NAME.getMethodName() + "-manual", 16010, System.currentTimeMillis())
          .toString();
  this.logName = AbstractFSWALProvider.getWALDirectoryName(serverName);
  this.logDir = new Path(walRootDir, logName);
  this.wals = new WALFactory(conf, TEST_NAME.getMethodName());

  Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
  TableDescriptor td = pair.getFirst();
  RegionInfo ri = pair.getSecond();

  WAL wal = createWAL(walFs, walRootDir, logName);
  HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
  writeData(td, region);

  // Now close the region without flush
  region.close(true);
  wal.shutdown();
  // split the log
  WALSplitter.split(walRootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);

  WAL wal2 = createWAL(walFs, walRootDir, logName);
  HRegion region2 = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal2);
  Result result2 = region2.get(new Get(ROW));
  assertEquals(td.getColumnFamilies().length, result2.size());
  for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
    assertTrue(Bytes.equals(VALUE1, result2.getValue(cfd.getName(), QUALIFIER)));
  }
}
 

private FSHLog createWAL() throws IOException {
  String logPrefix = factory.factoryId + WAL_FILE_NAME_DELIMITER + providerId;
  return new IOTestWAL(CommonFSUtils.getWALFileSystem(conf), CommonFSUtils.getWALRootDir(conf),
      AbstractFSWALProvider.getWALDirectoryName(factory.factoryId),
      HConstants.HREGION_OLDLOGDIR_NAME, conf, listeners, true, logPrefix,
      META_WAL_PROVIDER_ID.equals(providerId) ? META_WAL_PROVIDER_ID : null);
}
 

@Test
public void testSyncRunnerIndexOverflow() throws IOException, NoSuchFieldException,
    SecurityException, IllegalArgumentException, IllegalAccessException {
  final String name = this.name.getMethodName();
  FSHLog log = new FSHLog(FS, CommonFSUtils.getRootDir(CONF), name,
    HConstants.HREGION_OLDLOGDIR_NAME, CONF, null, true, null, null);
  log.init();
  try {
    Field ringBufferEventHandlerField = FSHLog.class.getDeclaredField("ringBufferEventHandler");
    ringBufferEventHandlerField.setAccessible(true);
    FSHLog.RingBufferEventHandler ringBufferEventHandler =
        (FSHLog.RingBufferEventHandler) ringBufferEventHandlerField.get(log);
    Field syncRunnerIndexField =
        FSHLog.RingBufferEventHandler.class.getDeclaredField("syncRunnerIndex");
    syncRunnerIndexField.setAccessible(true);
    syncRunnerIndexField.set(ringBufferEventHandler, Integer.MAX_VALUE - 1);
    TableDescriptor htd =
        TableDescriptorBuilder.newBuilder(TableName.valueOf(this.name.getMethodName()))
          .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
    NavigableMap<byte[], Integer> scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
    for (byte[] fam : htd.getColumnFamilyNames()) {
      scopes.put(fam, 0);
    }
    RegionInfo hri = RegionInfoBuilder.newBuilder(htd.getTableName()).build();
    MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
    for (int i = 0; i < 10; i++) {
      addEdits(log, hri, htd, 1, mvcc, scopes, "row");
    }
  } finally {
    log.close();
  }
}
 

@Test
public void testWALMonotonicallyIncreasingSeqId() throws Exception {
  List<Thread> putThreads = new ArrayList<>();
  for (int i = 0; i < 1; i++) {
    putThreads.add(new PutThread(region));
  }
  IncThread incThread = new IncThread(region);
  for (int i = 0; i < 1; i++) {
    putThreads.get(i).start();
  }
  incThread.start();
  incThread.join();

  Path logPath = ((AbstractFSWAL<?>) region.getWAL()).getCurrentFileName();
  region.getWAL().rollWriter();
  Thread.sleep(10);
  Path hbaseDir = new Path(walConf.get(HConstants.HBASE_DIR));
  Path oldWalsDir = new Path(hbaseDir, HConstants.HREGION_OLDLOGDIR_NAME);
  try (WAL.Reader reader = createReader(logPath, oldWalsDir)) {
    long currentMaxSeqid = 0;
    for (WAL.Entry e; (e = reader.next()) != null;) {
      if (!WALEdit.isMetaEditFamily(e.getEdit().getCells().get(0))) {
        long currentSeqid = e.getKey().getSequenceId();
        if (currentSeqid > currentMaxSeqid) {
          currentMaxSeqid = currentSeqid;
        } else {
          fail("Current max Seqid is " + currentMaxSeqid +
            ", but the next seqid in wal is smaller:" + currentSeqid);
        }
      }
    }
  }
}
 

public MockWAL(FileSystem fs, Path rootDir, String logName, Configuration conf)
    throws IOException {
  super(fs, rootDir, logName, HConstants.HREGION_OLDLOGDIR_NAME, conf, null, true, null, null);
}
 

public static boolean isArchivedLogFile(Path p) {
  String oldLog = Path.SEPARATOR + HConstants.HREGION_OLDLOGDIR_NAME + Path.SEPARATOR;
  return p.toString().contains(oldLog);
}
 

/**
 * @param walRootDir Path to the root directory where hbase files are stored
 * @param serverName Region Server owner of the log
 * @param logName WAL file name
 */
public WALLink(final Path walRootDir, final String serverName, final String logName) {
  final Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
  final Path logDir = new Path(new Path(walRootDir, HConstants.HREGION_LOGDIR_NAME), serverName);
  setLocations(new Path(logDir, logName), new Path(oldLogDir, logName));
}
 

/**
 * Constructor.
 * @param fs filesystem handle
 * @param root path for stored and archived wals
 * @param logDir dir where wals are stored
 * @param conf configuration to use
 */
@VisibleForTesting
public FSHLog(final FileSystem fs, final Path root, final String logDir, final Configuration conf)
    throws IOException {
  this(fs, root, logDir, HConstants.HREGION_OLDLOGDIR_NAME, conf, null, true, null, null);
}