Java 类org.apache.hadoop.hbase.wal.WAL 实例源码

@Override public WAL getWAL(HRegionInfo regionInfo) throws IOException {
  WAL wal;
  LogRoller roller = walRoller;
  //_ROOT_ and hbase:meta regions have separate WAL.
  if (regionInfo != null && regionInfo.isMetaTable()
      && regionInfo.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
    roller = ensureMetaWALRoller();
    wal = walFactory.getMetaWAL(regionInfo.getEncodedNameAsBytes());
  } else if (regionInfo == null) {
    wal = walFactory.getWAL(UNSPECIFIED_REGION);
  } else {
    wal = walFactory.getWAL(regionInfo.getEncodedNameAsBytes());
  }
  roller.addWAL(wal);
  return wal;
}

private static long writeMarker(final WAL wal, final HTableDescriptor htd, final HRegionInfo hri,
    final WALEdit edit, final MultiVersionConcurrencyControl mvcc, final boolean sync)
throws IOException {
  // TODO: Pass in current time to use?
  WALKey key =
    new HLogKey(hri.getEncodedNameAsBytes(), hri.getTable(), System.currentTimeMillis(), mvcc);
  // Add it to the log but the false specifies that we don't need to add it to the memstore
  long trx = MultiVersionConcurrencyControl.NONE;
  try {
    trx = wal.append(htd, hri, key, edit, false);
    if (sync) wal.sync(trx);
  } finally {
    // If you get hung here, is it a real WAL or a mocked WAL? If the latter, you need to
    // trip the latch that is inside in getWriteEntry up in your mock. See down in the append
    // called from onEvent in FSHLog.
    MultiVersionConcurrencyControl.WriteEntry we = key.getWriteEntry();
    if (mvcc != null && we != null) mvcc.complete(we);
  }
  return trx;
}

/**
 * Convenience method creating new HRegions. Used by createTable. The {@link WAL} for the created
 * region needs to be closed explicitly, if it is not null. Use {@link HRegion#getWAL()} to get
 * access.
 *
 * @param info       Info for region to create.
 * @param rootDir    Root directory for HBase instance
 * @param tableDir   table directory
 * @param wal        shared WAL
 * @param initialize - true to initialize the region
 * @param ignoreWAL  - true to skip generate new wal if it is null, mostly for createTable
 * @return new HRegion
 * @throws IOException
 */
public static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
    final Path tableDir, final Configuration conf, final HTableDescriptor hTableDescriptor,
    final WAL wal, final boolean initialize, final boolean ignoreWAL) throws IOException {
  LOG.info("creating HRegion " + info.getTable().getNameAsString() + " HTD == " + hTableDescriptor
      + " RootDir = " + rootDir + " Table name == " + info.getTable().getNameAsString());
  FileSystem fs = FileSystem.get(conf);
  HRegionFileSystem.createRegionOnFileSystem(conf, fs, tableDir, info);
  WAL effectiveWAL = wal;
  if (wal == null && !ignoreWAL) {
    // TODO HBASE-11983 There'll be no roller for this wal?
    // The WAL subsystem will use the default rootDir rather than the passed
    // in rootDir
    // unless I pass along via the conf.
    Configuration confForWAL = new Configuration(conf);
    confForWAL.set(HConstants.HBASE_DIR, rootDir.toString());
    effectiveWAL = (new WALFactory(confForWAL,
        Collections.<WALActionsListener>singletonList(new MetricsWAL()),
        "hregion-" + RandomStringUtils.randomNumeric(8))).getWAL(info.getEncodedNameAsBytes());
  }
  HRegion region =
      HRegion.newHRegion(tableDir, effectiveWAL, fs, conf, info, hTableDescriptor, null);
  if (initialize) region.initialize(null);
  return region;
}

public static void warmupHRegion(final HRegionInfo info, final HTableDescriptor htd,
    final WAL wal, final Configuration conf, final RegionServerServices rsServices,
    final CancelableProgressable reporter) throws IOException {

  if (info == null) throw new NullPointerException("Passed region info is null");

  if (LOG.isDebugEnabled()) {
    LOG.debug("HRegion.Warming up region: " + info);
  }

  Path rootDir = FSUtils.getRootDir(conf);
  Path tableDir = FSUtils.getTableDir(rootDir, info.getTable());

  FileSystem fs = null;
  if (rsServices != null) {
    fs = rsServices.getFileSystem();
  }
  if (fs == null) {
    fs = FileSystem.get(conf);
  }

  HRegion r = HRegion.newHRegion(tableDir, wal, fs, conf, info, htd, null);
  r.initializeWarmup(reporter);
}

/**
 * Test the case where the secondary region replica is not in reads enabled state because it is
 * waiting for a flush or region open marker from primary region. Replaying CANNOT_FLUSH
 * flush marker entry should restore the reads enabled status in the region and allow the reads
 * to continue.
 */
@Test
public void testReplayingFlushRequestRestoresReadsEnabledState() throws IOException {
  disableReads(secondaryRegion);

  // Test case 1: Test that replaying CANNOT_FLUSH request marker assuming this came from
  // triggered flush restores readsEnabled
  primaryRegion.flushcache(true, true);
  reader = createWALReaderForPrimary();
  while (true) {
    WAL.Entry entry = reader.next();
    if (entry == null) {
      break;
    }
    FlushDescriptor flush = WALEdit.getFlushDescriptor(entry.getEdit().getCells().get(0));
    if (flush != null) {
      secondaryRegion.replayWALFlushMarker(flush, entry.getKey().getLogSeqNum());
    }
  }

  // now reads should be enabled
  secondaryRegion.get(new Get(Bytes.toBytes(0)));
}

/**
 * Writes a marker to WAL indicating a flush is requested but cannot be complete due to various
 * reasons. Ignores exceptions from WAL. Returns whether the write succeeded.
 *
 * @param wal
 * @return whether WAL write was successful
 */
private boolean writeFlushRequestMarkerToWAL(WAL wal, boolean writeFlushWalMarker) {
  if (writeFlushWalMarker && wal != null && !writestate.readOnly) {
    FlushDescriptor desc = ProtobufUtil
        .toFlushDescriptor(FlushAction.CANNOT_FLUSH, getRegionInfo(), -1,
            new TreeMap<byte[], List<Path>>(Bytes.BYTES_COMPARATOR));
    try {
      WALUtil.writeFlushMarker(wal, this.htableDescriptor, getRegionInfo(), desc, true, mvcc);
      return true;
    } catch (IOException e) {
      LOG.warn(getRegionInfo().getEncodedName() + " : "
          + "Received exception while trying to write the flush request to wal", e);
    }
  }
  return false;
}

/**
 * Method to safely get the next sequence number.
 *
 * @return Next sequence number unassociated with any actual edit.
 * @throws IOException
 */
@VisibleForTesting protected long getNextSequenceId(final WAL wal) throws IOException {
  // TODO: For review. Putting an empty edit in to get a sequenceid out will
  // not work if the
  // WAL is banjaxed... if it has gotten an exception and the WAL has not yet
  // been rolled or
  // aborted. In this case, we'll just get stuck here. For now, until
  // HBASE-12751, just have
  // a timeout. May happen in tests after we tightened the semantic via
  // HBASE-14317.
  // Also, the getSequenceId blocks on a latch. There is no global list of
  // outstanding latches
  // so if an abort or stop, there is no way to call them in.
  WALKey key = this.appendEmptyEdit(wal);
  mvcc.complete(key.getWriteEntry());
  return key.getSequenceId(this.maxWaitForSeqId);
}

/**
 * Append a faked WALEdit in order to get a long sequence number and wal syncer will just ignore
 * the WALEdit append later.
 *
 * @param wal
 * @return Return the key used appending with no sync and no append.
 * @throws IOException
 */
private WALKey appendEmptyEdit(final WAL wal) throws IOException {
  // we use HLogKey here instead of WALKey directly to support legacy
  // coprocessors.
  @SuppressWarnings("deprecation") WALKey key =
      new HLogKey(getRegionInfo().getEncodedNameAsBytes(), getRegionInfo().getTable(),
          WALKey.NO_SEQUENCE_ID, 0, null, HConstants.NO_NONCE, HConstants.NO_NONCE, getMVCC());

  // Call append but with an empty WALEdit. The returned sequence id will not
  // be associated
  // with any edit and we can be sure it went in after all outstanding
  // appends.
  try {
    wal.append(getTableDesc(), getRegionInfo(), key, WALEdit.EMPTY_WALEDIT, false);
  } catch (Throwable t) {
    // If exception, our mvcc won't get cleaned up by client, so do it here.
    getMVCC().complete(key.getWriteEntry());
  }
  return key;
}

private void addWALEdits(final TableName tableName, final HRegionInfo hri, final byte[] rowName,
    final byte[] family, final int count, EnvironmentEdge ee, final WAL wal,
    final HTableDescriptor htd, final MultiVersionConcurrencyControl mvcc) throws IOException {
  String familyStr = Bytes.toString(family);
  long txid = -1;
  for (int j = 0; j < count; j++) {
    byte[] qualifierBytes = Bytes.toBytes(Integer.toString(j));
    byte[] columnBytes = Bytes.toBytes(familyStr + ":" + Integer.toString(j));
    WALEdit edit = new WALEdit();
    edit.add(new KeyValue(rowName, family, qualifierBytes, ee.currentTime(), columnBytes));
    // uses WALKey instead of HLogKey on purpose. will only work for tests where we don't care
    // about legacy coprocessors
    txid = wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName,
        ee.currentTime(), mvcc), edit, true);
  }
  if (-1 != txid) {
    wal.sync(txid);
  }
}

private HRegion mergeAndVerify(final String msg, final String regionName1,
  final String regionName2, final WAL log, final int upperbound)
throws Exception {
  Merge merger = new Merge(this.conf);
  LOG.info(msg);
  LOG.info("fs2=" + this.conf.get("fs.defaultFS"));
  int errCode = ToolRunner.run(this.conf, merger,
    new String[] {this.desc.getTableName().getNameAsString(), regionName1, regionName2}
  );
  assertTrue("'" + msg + "' failed with errCode " + errCode, errCode == 0);
  HRegionInfo mergedInfo = merger.getMergedHRegionInfo();

  // Now verify that we can read all the rows from regions 0, 1
  // in the new merged region.
  HRegion merged = HRegion.openHRegion(mergedInfo, this.desc, log, this.conf);
  verifyMerge(merged, upperbound);
  merged.close();
  LOG.info("Verified " + msg);
  return merged;
}

/**
 * Utility method to setup a WAL mock.
 * Needs to do the bit where we close latch on the WALKey on append else test hangs.
 * @return
 * @throws IOException
 */
private WAL mockWAL() throws IOException {
  WAL wal = mock(WAL.class);
  Mockito.when(wal.append((HTableDescriptor)Mockito.any(), (HRegionInfo)Mockito.any(),
      (WALKey)Mockito.any(), (WALEdit)Mockito.any(), Mockito.anyBoolean())).
    thenAnswer(new Answer<Long>() {
      @Override
      public Long answer(InvocationOnMock invocation) throws Throwable {
        WALKey key = invocation.getArgumentAt(2, WALKey.class);
        MultiVersionConcurrencyControl.WriteEntry we = key.getMvcc().begin();
        key.setWriteEntry(we);
        return 1L;
      }

  });
  return wal;
}

/**
 * Atomic bulk load.
 */
@Test
public void testAtomicBulkLoad() throws Exception {
  TableName TABLE_NAME = TableName.valueOf("atomicBulkLoad");

  int millisToRun = 30000;
  int numScanners = 50;

  UTIL.startMiniCluster(1);
  try {
    WAL log = UTIL.getHBaseCluster().getRegionServer(0).getWAL(null);
    FindBulkHBaseListener listener = new FindBulkHBaseListener();
    log.registerWALActionsListener(listener);
    runAtomicBulkloadTest(TABLE_NAME, millisToRun, numScanners);
    assertThat(listener.isFound(), is(true));
  } finally {
    UTIL.shutdownMiniCluster();
  }
}

private void addWALEdits(final TableName tableName, final HRegionInfo hri, final byte[] rowName,
    final byte[] family, final int count, EnvironmentEdge ee, final WAL wal,
    final HTableDescriptor htd, final MultiVersionConcurrencyControl mvcc)
throws IOException {
  String familyStr = Bytes.toString(family);
  for (int j = 0; j < count; j++) {
    byte[] qualifierBytes = Bytes.toBytes(Integer.toString(j));
    byte[] columnBytes = Bytes.toBytes(familyStr + ":" + Integer.toString(j));
    WALEdit edit = new WALEdit();
    edit.add(new KeyValue(rowName, family, qualifierBytes,
      ee.currentTime(), columnBytes));
    wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName,999, mvcc),
        edit, true);
  }
  wal.sync();
}

protected void addEdits(WAL log,
                        HRegionInfo hri,
                        HTableDescriptor htd,
                        int times,
                        MultiVersionConcurrencyControl mvcc)
    throws IOException {
  final byte[] row = Bytes.toBytes("row");
  for (int i = 0; i < times; i++) {
    long timestamp = System.currentTimeMillis();
    WALEdit cols = new WALEdit();
    cols.add(new KeyValue(row, row, row, timestamp, row));
    WALKey key = new WALKey(hri.getEncodedNameAsBytes(), htd.getTableName(),
        WALKey.NO_SEQUENCE_ID, timestamp, WALKey.EMPTY_UUIDS, HConstants.NO_NONCE,
        HConstants.NO_NONCE, mvcc);
    log.append(htd, hri, key, cols, true);
  }
  log.sync();
}

/**
 * Tests that the LogRoller perform the roll even if there are no edits
 */
@Test
public void testNoEdits() throws Exception {
  TableName tableName = TableName.valueOf("TestLogRollPeriodNoEdits");
  TEST_UTIL.createTable(tableName, "cf");
  try {
    Table table = new HTable(TEST_UTIL.getConfiguration(), tableName);
    try {
      HRegionServer server = TEST_UTIL.getRSForFirstRegionInTable(tableName);
      WAL log = server.getWAL(null);
      checkMinLogRolls(log, 5);
    } finally {
      table.close();
    }
  } finally {
    TEST_UTIL.deleteTable(tableName);
  }
}

private void checkMinLogRolls(final WAL log, final int minRolls)
    throws Exception {
  final List<Path> paths = new ArrayList<Path>();
  log.registerWALActionsListener(new WALActionsListener.Base() {
    @Override
    public void postLogRoll(Path oldFile, Path newFile) {
      LOG.debug("postLogRoll: oldFile="+oldFile+" newFile="+newFile);
      paths.add(newFile);
    }
  });

  // Sleep until we should get at least min-LogRoll events
  long wtime = System.currentTimeMillis();
  Thread.sleep((minRolls + 1) * LOG_ROLL_PERIOD);
  // Do some extra sleep in case the machine is slow,
  // and the log-roll is not triggered exactly on LOG_ROLL_PERIOD.
  final int NUM_RETRIES = 1 + 8 * (minRolls - paths.size());
  for (int retry = 0; paths.size() < minRolls && retry < NUM_RETRIES; ++retry) {
    Thread.sleep(LOG_ROLL_PERIOD / 4);
  }
  wtime = System.currentTimeMillis() - wtime;
  LOG.info(String.format("got %d rolls after %dms (%dms each) - expected at least %d rolls",
                         paths.size(), wtime, wtime / paths.size(), minRolls));
  assertFalse(paths.size() < minRolls);
}

/**
 * Tests that logs are deleted
 * @throws IOException
 * @throws org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException
 */
@Test
public void testLogRolling() throws Exception {
  this.tableName = getName();
    // TODO: Why does this write data take for ever?
    startAndWriteData();
  final WAL log = server.getWAL(null);
  LOG.info("after writing there are " + DefaultWALProvider.getNumRolledLogFiles(log) +
      " log files");

    // flush all regions
    for (Region r: server.getOnlineRegionsLocalContext()) {
      r.flush(true);
    }

    // Now roll the log
    log.rollWriter();

  int count = DefaultWALProvider.getNumRolledLogFiles(log);
  LOG.info("after flushing all regions and rolling logs there are " + count + " log files");
    assertTrue(("actual count: " + count), count <= 2);
}

@Test
public void testIncrementWithReturnResultsSetToFalse() throws Exception {
  byte[] row1 = Bytes.toBytes("row1");
  byte[] col1 = Bytes.toBytes("col1");

  // Setting up region
  final WALFactory wals = new WALFactory(CONF, null, "testIncrementWithReturnResultsSetToFalse");
  byte[] tableName = Bytes.toBytes("testIncrementWithReturnResultsSetToFalse");
  final WAL wal = wals.getWAL(tableName);
  HRegion region = createHRegion(tableName, "increment", wal, Durability.USE_DEFAULT);

  Increment inc1 = new Increment(row1);
  inc1.setReturnResults(false);
  inc1.addColumn(FAMILY, col1, 1);
  Result res = region.increment(inc1);
  assertNull(res);
}

private HRegion createHRegion (byte [] tableName, String callingMethod,
  WAL log, Durability durability)
throws IOException {
  HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
  htd.setDurability(durability);
  HColumnDescriptor hcd = new HColumnDescriptor(FAMILY);
  htd.addFamily(hcd);
  HRegionInfo info = new HRegionInfo(htd.getTableName(), null, null, false);
  Path path = new Path(DIR + callingMethod);
  if (FS.exists(path)) {
    if (!FS.delete(path, true)) {
      throw new IOException("Failed delete of " + path);
    }
  }
  return HRegion.createHRegion(info, path, CONF, htd, log);
}

private HRegion getRegion(final Configuration conf, final String tableName) throws IOException {
  WAL wal = new FSHLog(FileSystem.get(conf), TEST_UTIL.getDataTestDir(),
    TEST_UTIL.getDataTestDir().toString(), conf);
  return (HRegion)TEST_UTIL.createLocalHRegion(Bytes.toBytes(tableName),
    HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, tableName, conf,
    false, Durability.SKIP_WAL, wal, INCREMENT_BYTES);
}

/**
 * @return the WAL associated with the given region
 * @throws IOException e
 */
public synchronized WAL getLog(HRegionInfo info) throws IOException {
  if (this.walFactory == null) {
    String logName = 
        HConstants.HREGION_LOGDIR_NAME + "_" + System.currentTimeMillis();
    final Configuration walConf = new Configuration(this.conf);
    FSUtils.setRootDir(walConf, fs.getHomeDirectory());
    this.walFactory = new WALFactory(walConf, null, logName);
  }
  final byte[] region = info.getEncodedNameAsBytes();
  return info.isMetaRegion() ? walFactory.getMetaWAL(region) : walFactory.getWAL(region);
}

/**
 * Write the marker that a compaction has succeeded and is about to be committed.
 * This provides info to the HMaster to allow it to recover the compaction if
 * this regionserver dies in the middle (This part is not yet implemented). It also prevents
 * the compaction from finishing if this regionserver has already lost its lease on the log.
 * @param mvcc Used by WAL to get sequence Id for the waledit.
 */
public static long writeCompactionMarker(WAL wal, HTableDescriptor htd, HRegionInfo hri,
    final CompactionDescriptor c, MultiVersionConcurrencyControl mvcc)
throws IOException {
  long trx = writeMarker(wal, htd, hri, WALEdit.createCompaction(hri, c), mvcc, true);
  if (LOG.isTraceEnabled()) {
    LOG.trace("Appended compaction marker " + TextFormat.shortDebugString(c));
  }
  return trx;
}

/**
 * Write a flush marker indicating a start / abort or a complete of a region flush
 */
public static long writeFlushMarker(WAL wal, HTableDescriptor htd, HRegionInfo hri,
    final FlushDescriptor f, boolean sync, MultiVersionConcurrencyControl mvcc)
throws IOException {
  long trx = writeMarker(wal, htd, hri, WALEdit.createFlushWALEdit(hri, f), mvcc, sync);
  if (LOG.isTraceEnabled()) {
    LOG.trace("Appended flush marker " + TextFormat.shortDebugString(f));
  }
  return trx;
}

/**
 * Write a region open marker indicating that the region is opened
 */
public static long writeRegionEventMarker(WAL wal, HTableDescriptor htd, HRegionInfo hri,
    final RegionEventDescriptor r, final MultiVersionConcurrencyControl mvcc)
throws IOException {
  long trx = writeMarker(wal, htd, hri, WALEdit.createRegionEventWALEdit(hri, r), mvcc, true);
  if (LOG.isTraceEnabled()) {
    LOG.trace("Appended region event marker " + TextFormat.shortDebugString(r));
  }
  return trx;
}

/**
 * Write a log marker that a bulk load has succeeded and is about to be committed.
 *
 * @param wal        The log to write into.
 * @param htd        A description of the table that we are bulk loading into.
 * @param hri       A description of the region in the table that we are bulk loading into.
 * @param desc A protocol buffers based description of the client's bulk loading request
 * @return txid of this transaction or if nothing to do, the last txid
 * @throws IOException We will throw an IOException if we can not append to the HLog.
 */
public static long writeBulkLoadMarkerAndSync(final WAL wal, final HTableDescriptor htd,
    final HRegionInfo hri, final WALProtos.BulkLoadDescriptor desc,
    final MultiVersionConcurrencyControl mvcc)
throws IOException {
  long trx = writeMarker(wal, htd, hri, WALEdit.createBulkLoadEvent(hri, desc), mvcc, true);
  if (LOG.isTraceEnabled()) {
    LOG.trace("Appended Bulk Load marker " + TextFormat.shortDebugString(desc));
  }
  return trx;
}

/**
 * Constructor
 * @param implClass - not used
 * @param impl the coprocessor instance
 * @param priority chaining priority
 * @param seq load sequence
 * @param conf configuration
 * @param wal WAL
 */
public WALEnvironment(Class<?> implClass, final Coprocessor impl,
    final int priority, final int seq, final Configuration conf,
    final WAL wal) {
  super(impl, priority, seq, conf);
  this.wal = wal;
  // Pick which version of the API we'll call.
  // This way we avoid calling the new version on older WALObservers so
  // we can maintain binary compatibility.
  // See notes in javadoc for WALObserver
  useLegacyPre = useLegacyMethod(impl.getClass(), "preWALWrite", ObserverContext.class,
      HRegionInfo.class, WALKey.class, WALEdit.class);
  useLegacyPost = useLegacyMethod(impl.getClass(), "postWALWrite", ObserverContext.class,
      HRegionInfo.class, WALKey.class, WALEdit.class);
}

/**
 * Constructor
 * @param log the write ahead log
 * @param conf the configuration
 */
public WALCoprocessorHost(final WAL log, final Configuration conf) {
  // We don't want to require an Abortable passed down through (FS)HLog, so
  // this means that a failure to load of a WAL coprocessor won't abort the
  // server. This isn't ideal, and means that security components that
  // utilize a WALObserver will have to check the observer initialization
  // state manually. However, WALObservers will eventually go away so it
  // should be an acceptable state of affairs.
  super(null);
  this.wal = log;
  // load system default cp's from configuration.
  loadSystemCoprocessors(conf, WAL_COPROCESSOR_CONF_KEY);
}

private static void transformFile(Path input, Path output)
    throws IOException {
  Configuration conf = HBaseConfiguration.create();

  FileSystem inFS = input.getFileSystem(conf);
  FileSystem outFS = output.getFileSystem(conf);

  WAL.Reader in = WALFactory.createReaderIgnoreCustomClass(inFS, input, conf);
  WALProvider.Writer out = null;

  try {
    if (!(in instanceof ReaderBase)) {
      System.err.println("Cannot proceed, invalid reader type: " + in.getClass().getName());
      return;
    }
    boolean compress = ((ReaderBase)in).hasCompression();
    conf.setBoolean(HConstants.ENABLE_WAL_COMPRESSION, !compress);
    out = WALFactory.createWALWriter(outFS, output, conf);

    WAL.Entry e = null;
    while ((e = in.next()) != null) out.append(e);
  } finally {
    in.close();
    if (out != null) {
      out.close();
      out = null;
    }
  }
}

public void requestRollAll() {
  for (WAL wal : walNeedsRoll.keySet()) {
    walNeedsRoll.put(wal, Boolean.TRUE);
  }
  synchronized(rollLog) {
    rollLog.set(true);
    rollLog.notifyAll();
  }
}

/**
 * Test the case where the secondary region replica is not in reads enabled state because it is
 * waiting for a flush or region open marker from primary region. Replaying region open event
 * entry from primary should restore the reads enabled status in the region and allow the reads
 * to continue.
 */
@Test
public void testReplayingRegionOpenEventRestoresReadsEnabledState() throws IOException {
  // Test case 3: Test that replaying region open event markers restores readsEnabled
  disableReads(secondaryRegion);

  primaryRegion.close();
  primaryRegion = HRegion.openHRegion(rootDir, primaryHri, htd, walPrimary, CONF, rss, null);

  reader = createWALReaderForPrimary();
  while (true) {
    WAL.Entry entry = reader.next();
    if (entry == null) {
      break;
    }

    RegionEventDescriptor regionEventDesc
      = WALEdit.getRegionEventDescriptor(entry.getEdit().getCells().get(0));

    if (regionEventDesc != null) {
      secondaryRegion.replayWALRegionEventMarker(regionEventDesc);
    }
  }

  // now reads should be enabled
  secondaryRegion.get(new Get(Bytes.toBytes(0)));
}

private static void processTable(final FileSystem fs, final Path p, final WALFactory walFactory,
    final Configuration c, final boolean majorCompact) throws IOException {
  HRegion region;
  FSTableDescriptors fst = new FSTableDescriptors(c);
  // Currently expects tables have one region only.
  if (FSUtils.getTableName(p).equals(TableName.META_TABLE_NAME)) {
    final WAL wal =
        walFactory.getMetaWAL(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes());
    region = HRegion.newHRegion(p, wal, fs, c, HRegionInfo.FIRST_META_REGIONINFO,
        fst.get(TableName.META_TABLE_NAME), null);
  } else {
    throw new IOException("Not a known catalog table: " + p.toString());
  }
  try {
    region.mvcc.advanceTo(region.initialize(null));
    if (majorCompact) {
      region.compact(true);
    } else {
      // Default behavior
      Scan scan = new Scan();
      // scan.addFamily(HConstants.CATALOG_FAMILY);
      RegionScanner scanner = region.getScanner(scan);
      try {
        List<Cell> kvs = new ArrayList<Cell>();
        boolean done;
        do {
          kvs.clear();
          done = scanner.next(kvs);
          if (kvs.size() > 0) LOG.info(kvs);
        } while (done);
      } finally {
        scanner.close();
      }
    }
  } finally {
    region.close();
  }
}

private void writeRegionOpenMarker(WAL wal, long openSeqId) throws IOException {
  Map<byte[], List<Path>> storeFiles = getStoreFiles();
  RegionEventDescriptor regionOpenDesc = ProtobufUtil
      .toRegionEventDescriptor(RegionEventDescriptor.EventType.REGION_OPEN, getRegionInfo(),
          openSeqId, getRegionServerServices().getServerName(), storeFiles);
  WALUtil.writeRegionEventMarker(wal, getTableDesc(), getRegionInfo(), regionOpenDesc, mvcc);
}

@After
public void teardown() throws IOException {
  if (region != null) {
    BlockCache bc = region.getStores().get(0).getCacheConfig().getBlockCache();
    ((HRegion)region).close();
    WAL wal = ((HRegion)region).getWAL();
    if (wal != null) wal.close();
    if (bc != null) bc.shutdown();
    region = null;
  }
}

/**
 * Sanity check that we can move logs around while we are reading
 * from them. Should this test fail, ReplicationSource would have a hard
 * time reading logs that are being archived.
 * @throws Exception
 */
@Test
public void testLogMoving() throws Exception{
  Path logPath = new Path(logDir, "log");
  if (!FS.exists(logDir)) FS.mkdirs(logDir);
  if (!FS.exists(oldLogDir)) FS.mkdirs(oldLogDir);
  WALProvider.Writer writer = WALFactory.createWALWriter(FS, logPath,
      TEST_UTIL.getConfiguration());
  for(int i = 0; i < 3; i++) {
    byte[] b = Bytes.toBytes(Integer.toString(i));
    KeyValue kv = new KeyValue(b,b,b);
    WALEdit edit = new WALEdit();
    edit.add(kv);
    WALKey key = new WALKey(b, TableName.valueOf(b), 0, 0,
        HConstants.DEFAULT_CLUSTER_ID);
    writer.append(new WAL.Entry(key, edit));
    writer.sync();
  }
  writer.close();

  WAL.Reader reader = WALFactory.createReader(FS, logPath, TEST_UTIL.getConfiguration());
  WAL.Entry entry = reader.next();
  assertNotNull(entry);

  Path oldLogPath = new Path(oldLogDir, "log");
  FS.rename(logPath, oldLogPath);

  entry = reader.next();
  assertNotNull(entry);

  entry = reader.next();
  entry = reader.next();

  assertNull(entry);
  reader.close();
}

/**
 * Tests the case where a request for flush cache is sent to the region, but region cannot flush.
 * It should write the flush request marker instead.
 */
@Test
public void testWriteFlushRequestMarker() throws IOException {
  // primary region is empty at this point. Request a flush with writeFlushRequestWalMarker=false
  FlushResultImpl result = (FlushResultImpl)((HRegion)primaryRegion).flushcache(true, false);
  assertNotNull(result);
  assertEquals(result.result, FlushResultImpl.Result.CANNOT_FLUSH_MEMSTORE_EMPTY);
  assertFalse(result.wroteFlushWalMarker);

  // request flush again, but this time with writeFlushRequestWalMarker = true
  result = (FlushResultImpl)((HRegion)primaryRegion).flushcache(true, true);
  assertNotNull(result);
  assertEquals(result.result, FlushResultImpl.Result.CANNOT_FLUSH_MEMSTORE_EMPTY);
  assertTrue(result.wroteFlushWalMarker);

  List<FlushDescriptor> flushes = Lists.newArrayList();
  reader = createWALReaderForPrimary();
  while (true) {
    WAL.Entry entry = reader.next();
    if (entry == null) {
      break;
    }
    FlushDescriptor flush = WALEdit.getFlushDescriptor(entry.getEdit().getCells().get(0));
    if (flush != null) {
      flushes.add(flush);
    }
  }

  assertEquals(1, flushes.size());
  assertNotNull(flushes.get(0));
  assertEquals(FlushDescriptor.FlushAction.CANNOT_FLUSH, flushes.get(0).getAction());
}

/**
 * Create an unmanaged WAL. Be sure to close it when you're through.
 */
public static WAL createWal(final Configuration conf, final Path rootDir, final HRegionInfo hri)
    throws IOException {
  // The WAL subsystem will use the default rootDir rather than the passed in rootDir
  // unless I pass along via the conf.
  Configuration confForWAL = new Configuration(conf);
  confForWAL.set(HConstants.HBASE_DIR, rootDir.toString());
  return (new WALFactory(confForWAL,
      Collections.<WALActionsListener>singletonList(new MetricsWAL()),
      "hregion-" + RandomStringUtils.randomNumeric(8))).
      getWAL(hri.getEncodedNameAsBytes());
}

/**
 * Create a region with it's own WAL. Be sure to call
 * {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} to clean up all resources.
 */
public static HRegion createRegionAndWAL(final HRegionInfo info, final Path rootDir,
    final Configuration conf, final HTableDescriptor htd, boolean initialize)
    throws IOException {
  WAL wal = createWal(conf, rootDir, info);
  return HRegion.createHRegion(info, rootDir, conf, htd, wal, initialize);
}

static int replayEdit(HRegion region, WAL.Entry entry) throws IOException {
  if (WALEdit.isMetaEditFamily(entry.getEdit().getCells().get(0))) {
    return 0; // handled elsewhere
  }
  Put put = new Put(entry.getEdit().getCells().get(0).getRow());
  for (Cell cell : entry.getEdit().getCells()) put.add(cell);
  put.setDurability(Durability.SKIP_WAL);
  MutationReplay mutation = new MutationReplay(MutationType.PUT, put, 0, 0);
  region.batchReplay(new MutationReplay[] {mutation},
    entry.getKey().getLogSeqNum());
  return Integer.parseInt(Bytes.toString(put.getRow()));
}

/**
 * @return A region on which you must call
 *         {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} when done.
 */
public HRegion initHRegion(TableName tableName, byte[] startKey, byte[] stopKey, WAL wal)
throws IOException {
  return TEST_UTIL.createLocalHRegion(tableName.getName(), startKey, stopKey,
    getName(), CONF, false, Durability.SYNC_WAL,
    wal, COLUMN_FAMILY_BYTES);
}

/**
 * @return A region on which you must call
 *         {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)} when done.
 */
public HRegion initHRegion(TableName tableName, byte[] startKey, byte[] stopKey, WAL wal)
throws IOException {
  return TEST_UTIL.createLocalHRegion(tableName.getName(), startKey, stopKey,
    getName(), CONF, false, Durability.SYNC_WAL,
    wal, COLUMN_FAMILY_BYTES);
}