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

@Override
public void postProcess(HRegion region, WALEdit walEdit, boolean success) throws IOException {
  RegionCoprocessorHost coprocessorHost = region.getCoprocessorHost();
  if (coprocessorHost != null) {
    for (Mutation m : mutations) {
      if (m instanceof Put) {
        coprocessorHost.postPut((Put) m, walEdit, m.getDurability());
      } else if (m instanceof Delete) {
        coprocessorHost.postDelete((Delete) m, walEdit, m.getDurability());
      }
    }
    // At the end call the CP hook postBatchMutateIndispensably
    if (miniBatch != null) {
      // Directly calling this hook, with out calling pre/postBatchMutate() when Processor do a
      // read only process. Then no need to call this batch based CP hook also.
      coprocessorHost.postBatchMutateIndispensably(miniBatch, success);
    }
  }
}

/**
 * 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;
}

@Override
public void map(WALKey key, WALEdit value,
  Context context)
throws IOException {
  try {
    // skip all other tables
    if (Bytes.equals(table, key.getTablename().getName())) {
      for (Cell cell : value.getCells()) {
        KeyValue kv = KeyValueUtil.ensureKeyValueTypeForMR(cell);
        if (WALEdit.isMetaEditFamily(kv.getFamily())) continue;
        context.write(new ImmutableBytesWritable(kv.getRow()), kv);
      }
    }
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
}

/**
 * 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)));
}

@Override
protected boolean matchesSafely(WALEdit item) {
  assertTrue(Arrays.equals(item.getCells().get(0).getQualifier(), typeBytes));
  BulkLoadDescriptor desc;
  try {
    desc = WALEdit.getBulkLoadDescriptor(item.getCells().get(0));
  } catch (IOException e) {
    return false;
  }
  assertNotNull(desc);

  if (tableName != null) {
    assertTrue(Bytes.equals(ProtobufUtil.toTableName(desc.getTableName()).getName(),
      tableName));
  }

  if(storeFileNames != null) {
    int index=0;
    StoreDescriptor store = desc.getStores(0);
    assertTrue(Bytes.equals(store.getFamilyName().toByteArray(), familyName));
    assertTrue(Bytes.equals(Bytes.toBytes(store.getStoreHomeDir()), familyName));
    assertEquals(storeFileNames.size(), store.getStoreFileCount());
  }

  return true;
}

/**
 * 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;
}

/**
 * @param info
 * @param logKey
 * @param logEdit
 * @return true if default behavior should be bypassed, false otherwise
 * @throws IOException
 */
public boolean preWALRestore(final HRegionInfo info, final WALKey logKey,
    final WALEdit logEdit) throws IOException {
  return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      // Once we don't need to support the legacy call, replace RegionOperation with a version
      // that's ObserverContext<RegionEnvironment> and avoid this cast.
      final RegionEnvironment env = (RegionEnvironment)ctx.getEnvironment();
      if (env.useLegacyPre) {
        if (logKey instanceof HLogKey) {
          oserver.preWALRestore(ctx, info, (HLogKey)logKey, logEdit);
        } else {
          legacyWarning(oserver.getClass(), "There are wal keys present that are not HLogKey.");
        }
      } else {
        oserver.preWALRestore(ctx, info, logKey, logEdit);
      }
    }
  });
}

/**
 * @param info
 * @param logKey
 * @param logEdit
 * @throws IOException
 */
public void postWALRestore(final HRegionInfo info, final WALKey logKey, final WALEdit logEdit)
    throws IOException {
  execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      // Once we don't need to support the legacy call, replace RegionOperation with a version
      // that's ObserverContext<RegionEnvironment> and avoid this cast.
      final RegionEnvironment env = (RegionEnvironment)ctx.getEnvironment();
      if (env.useLegacyPost) {
        if (logKey instanceof HLogKey) {
          oserver.postWALRestore(ctx, info, (HLogKey)logKey, logEdit);
        } else {
          legacyWarning(oserver.getClass(), "There are wal keys present that are not HLogKey.");
        }
      } else {
        oserver.postWALRestore(ctx, info, logKey, logEdit);
      }
    }
  });
}

@Override
public boolean matches(Object edit) {
  List<Cell> cells = ((WALEdit)edit).getCells();
  if (cells.isEmpty()) {
    return false;
  }
  if (WALEdit.isMetaEditFamily(cells.get(0))) {
    FlushDescriptor desc = null;
    try {
      desc = WALEdit.getFlushDescriptor(cells.get(0));
    } catch (IOException e) {
      LOG.warn(e);
      return false;
    }
    if (desc != null) {
      for (FlushAction action : actions) {
        if (desc.getAction() == action) {
          return true;
        }
      }
    }
  }
  return false;
}

/**
 * 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());
}

/**
 * Count the number of different row keys in the given edit because of mini-batching. We assume
 * that there's at least one Cell in the WALEdit.
 * @param edit edit to count row keys from
 * @return number of different row keys
 */
private int countDistinctRowKeys(WALEdit edit) {
  List<Cell> cells = edit.getCells();
  int distinctRowKeys = 1;
  Cell lastCell = cells.get(0);
  for (int i = 0; i < edit.size(); i++) {
    if (!CellUtil.matchingRow(cells.get(i), lastCell)) {
      distinctRowKeys++;
    }
  }
  return distinctRowKeys;
}

@SuppressWarnings("null")
@Override
public void prePut(final ObserverContext<RegionCoprocessorEnvironment> c,
                   final Put put, final WALEdit edit,
                   final Durability durability) {
  String tableName =
      c.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
  if (tableName.equals("observed_table")) {
    // Trigger a NPE to fail the coprocessor
    Integer i = null;
    i = i + 1;
  }
}

@Override
public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e,
    final Put put, final WALEdit edit, final Durability durability)
    throws IOException {
  Map<byte[], List<Cell>> familyMap = put.getFamilyCellMap();
  if (familyMap.containsKey(test)) {
    e.bypass();
  }
}

/**
 * Tests a case where we replay only a flush start marker, then the region is closed. This region
 * should not block indefinitely
 */
@Test (timeout = 60000)
public void testOnlyReplayingFlushStartDoesNotHoldUpRegionClose() throws IOException {
  // load some data to primary and flush
  int start = 0;
  LOG.info("-- Writing some data to primary from " +  start + " to " + (start+100));
  putData(primaryRegion, Durability.SYNC_WAL, start, 100, cq, families);
  LOG.info("-- Flushing primary, creating 3 files for 3 stores");
  primaryRegion.flush(true);

  // now replay the edits and the flush marker
  reader = createWALReaderForPrimary();

  LOG.info("-- Replaying edits and flush events in secondary");
  while (true) {
    WAL.Entry entry = reader.next();
    if (entry == null) {
      break;
    }
    FlushDescriptor flushDesc
      = WALEdit.getFlushDescriptor(entry.getEdit().getCells().get(0));
    if (flushDesc != null) {
      if (flushDesc.getAction() == FlushAction.START_FLUSH) {
        LOG.info("-- Replaying flush start in secondary");
        PrepareFlushResult result = secondaryRegion.replayWALFlushStartMarker(flushDesc);
      } else if (flushDesc.getAction() == FlushAction.COMMIT_FLUSH) {
        LOG.info("-- NOT Replaying flush commit in secondary");
      }
    } else {
      replayEdit(secondaryRegion, entry);
    }
  }

  assertTrue(rss.getRegionServerAccounting().getGlobalMemstoreSize() > 0);
  // now close the region which should not cause hold because of un-committed flush
  secondaryRegion.close();

  // verify that the memstore size is back to what it was
  assertEquals(0, rss.getRegionServerAccounting().getGlobalMemstoreSize());
}

@Override
public void postPut(final ObserverContext<RegionCoprocessorEnvironment> c,
    final Put put, final WALEdit edit, final Durability durability) {
  if (aclRegion) {
    updateACL(c.getEnvironment(), put.getFamilyCellMap());
  }
}

@Override
public void preDelete(final ObserverContext<RegionCoprocessorEnvironment> c,
    final Delete delete, final WALEdit edit, final Durability durability)
    throws IOException {
  // An ACL on a delete is useless, we shouldn't allow it
  if (delete.getAttribute(AccessControlConstants.OP_ATTRIBUTE_ACL) != null) {
    throw new DoNotRetryIOException("ACL on delete has no effect: " + delete.toString());
  }
  // Require WRITE permissions on all cells covered by the delete. Unlike
  // for Puts we need to check all visible prior versions, because a major
  // compaction could remove them. If the user doesn't have permission to
  // overwrite any of the visible versions ('visible' defined as not covered
  // by a tombstone already) then we have to disallow this operation.
  RegionCoprocessorEnvironment env = c.getEnvironment();
  Map<byte[],? extends Collection<Cell>> families = delete.getFamilyCellMap();
  User user = getActiveUser();
  AuthResult authResult = permissionGranted(OpType.DELETE, user, env, families, Action.WRITE);
  logResult(authResult);
  if (!authResult.isAllowed()) {
    if (cellFeaturesEnabled && !compatibleEarlyTermination) {
      delete.setAttribute(CHECK_COVERING_PERM, TRUE);
    } else if (authorizationEnabled) {
      throw new AccessDeniedException("Insufficient permissions " +
        authResult.toContextString());
    }
  }
}

@Override
public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> c,
    final Delete delete, final WALEdit edit, final Durability durability)
    throws IOException {
  if (aclRegion) {
    updateACL(c.getEnvironment(), delete.getFamilyCellMap());
  }
}

private void filterCellByStore(Entry logEntry) {
  Map<byte[], Long> maxSeqIdInStores =
      regionMaxSeqIdInStores.get(Bytes.toString(logEntry.getKey().getEncodedRegionName()));
  if (maxSeqIdInStores == null || maxSeqIdInStores.isEmpty()) {
    return;
  }
  // Create the array list for the cells that aren't filtered.
  // We make the assumption that most cells will be kept.
  ArrayList<Cell> keptCells = new ArrayList<Cell>(logEntry.getEdit().getCells().size());
  for (Cell cell : logEntry.getEdit().getCells()) {
    if (CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
      keptCells.add(cell);
    } else {
      byte[] family = CellUtil.cloneFamily(cell);
      Long maxSeqId = maxSeqIdInStores.get(family);
      // Do not skip cell even if maxSeqId is null. Maybe we are in a rolling upgrade,
      // or the master was crashed before and we can not get the information.
      if (maxSeqId == null || maxSeqId.longValue() < logEntry.getKey().getLogSeqNum()) {
        keptCells.add(cell);
      }
    }
  }

  // Anything in the keptCells array list is still live.
  // So rather than removing the cells from the array list
  // which would be an O(n^2) operation, we just replace the list
  logEntry.getEdit().setCells(keptCells);
}

/**
 * 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)));
}

/**
 * Coprocessors shouldn't get notice of empty waledits.
 */
@Test
public void testEmptyWALEditAreNotSeen() throws Exception {
  final HRegionInfo hri = createBasic3FamilyHRegionInfo(Bytes.toString(TEST_TABLE));
  final HTableDescriptor htd = createBasic3FamilyHTD(Bytes.toString(TEST_TABLE));
  final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();

  WAL log = wals.getWAL(UNSPECIFIED_REGION);
  try {
    SampleRegionWALObserver cp = getCoprocessor(log, SampleRegionWALObserver.class);

    cp.setTestValues(TEST_TABLE, null, null, null, null, null, null, null);

    assertFalse(cp.isPreWALWriteCalled());
    assertFalse(cp.isPostWALWriteCalled());

    final long now = EnvironmentEdgeManager.currentTime();
    long txid = log.append(htd, hri,
        new WALKey(hri.getEncodedNameAsBytes(), hri.getTable(), now, mvcc),
        new WALEdit(), true);
    log.sync(txid);

    assertFalse("Empty WALEdit should skip coprocessor evaluation.", cp.isPreWALWriteCalled());
    assertFalse("Empty WALEdit should skip coprocessor evaluation.", cp.isPostWALWriteCalled());
  } finally {
    log.close();
  }
}

@Override
public void visitLogEntryBeforeWrite(HTableDescriptor htd, WALKey logKey, WALEdit logEdit) {
  for (Cell cell : logEdit.getCells()) {
    KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
    for (Map.Entry entry : kv.toStringMap().entrySet()) {
      if (entry.getValue().equals(Bytes.toString(WALEdit.BULK_LOAD))) {
        found = true;
      }
    }
  }
}

@Override
public void preBatchMutate(HRegion region, WALEdit walEdit) throws IOException {
  // TODO we should return back the status of this hook run to HRegion so that those Mutations
  // with OperationStatus as SUCCESS or FAILURE should not get applied to memstore.
  RegionCoprocessorHost coprocessorHost = region.getCoprocessorHost();
  OperationStatus[] opStatus = new OperationStatus[mutations.size()];
  Arrays.fill(opStatus, OperationStatus.NOT_RUN);
  WALEdit[] walEditsFromCP = new WALEdit[mutations.size()];
  if (coprocessorHost != null) {
    miniBatch = new MiniBatchOperationInProgress<Mutation>(
        mutations.toArray(new Mutation[mutations.size()]), opStatus, walEditsFromCP, 0,
        mutations.size());
    coprocessorHost.preBatchMutate(miniBatch);
  }
  // Apply edits to a single WALEdit
  for (int i = 0; i < mutations.size(); i++) {
    if (opStatus[i] == OperationStatus.NOT_RUN) {
      // Other OperationStatusCode means that Mutation is already succeeded or failed in CP hook
      // itself. No need to apply again to region
      if (walEditsFromCP[i] != null) {
        // Add the WALEdit created by CP hook
        for (Cell walCell : walEditsFromCP[i].getCells()) {
          walEdit.add(walCell);
        }
      }
    }
  }
}

public MiniBatchOperationInProgress(T[] operations, OperationStatus[] retCodeDetails,
    WALEdit[] walEditsFromCoprocessors, int firstIndex, int lastIndexExclusive) {
  this.operations = operations;
  this.retCodeDetails = retCodeDetails;
  this.walEditsFromCoprocessors = walEditsFromCoprocessors;
  this.firstIndex = firstIndex;
  this.lastIndexExclusive = lastIndexExclusive;
}

/**
 * Append the given map of family->edits to a WALEdit data structure. This does not write to the
 * WAL itself.
 *
 * @param familyMap map of family->edits
 * @param walEdit   the destination entry to append into
 */
private void addFamilyMapToWALEdit(Map<byte[], List<Cell>> familyMap, WALEdit walEdit) {
  for (List<Cell> edits : familyMap.values()) {
    assert edits instanceof RandomAccess;
    int listSize = edits.size();
    for (int i = 0; i < listSize; i++) {
      Cell cell = edits.get(i);
      walEdit.add(cell);
    }
  }
}

/**
 * @param put The Put object
 * @param edit The WALEdit object.
 * @param durability The durability used
 * @return true if default processing should be bypassed
 * @exception IOException Exception
 */
public boolean prePut(final Put put, final WALEdit edit, final Durability durability)
    throws IOException {
  return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      oserver.prePut(ctx, put, edit, durability);
    }
  });
}

/**
 * @param delete The Delete object
 * @param edit The WALEdit object.
 * @param durability The durability used
 * @return true if default processing should be bypassed
 * @exception IOException Exception
 */
public boolean preDelete(final Delete delete, final WALEdit edit, final Durability durability)
    throws IOException {
  return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      oserver.preDelete(ctx, delete, edit, durability);
    }
  });
}

/**
 * @param delete The Delete object
 * @param edit The WALEdit object.
 * @param durability The durability used
 * @exception IOException Exception
 */
public void postDelete(final Delete delete, final WALEdit edit, final Durability durability)
    throws IOException {
  execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      oserver.postDelete(ctx, delete, edit, durability);
    }
  });
}

/**
 * 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();
}

/**
 * Test for HBASE-9038, Replication.scopeWALEdits would NPE if it wasn't filtering out
 * the compaction WALEdit
 * @throws Exception
 */
@Test(timeout=300000)
public void testCompactionWALEdits() throws Exception {
  WALProtos.CompactionDescriptor compactionDescriptor =
      WALProtos.CompactionDescriptor.getDefaultInstance();
  HRegionInfo hri = new HRegionInfo(htable1.getName(),
    HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
  WALEdit edit = WALEdit.createCompaction(hri, compactionDescriptor);
  Replication.scopeWALEdits(htable1.getTableDescriptor(), new WALKey(), edit);
}

private Entry createEntry(byte[]... kvs) {
  WALKey key1 = new WALKey(new byte[] {}, TableName.valueOf("foo"));
  WALEdit edit1 = new WALEdit();

  for (byte[] kv : kvs) {
    edit1.add(new KeyValue(kv, kv, kv));
  }
  return new Entry(key1, edit1);
}

private WALEdit getWALEdits(int count) {
  WALEdit edit = new WALEdit();
  for (int i = 0; i < count; i++) {
    edit.add(new KeyValue(Bytes.toBytes(System.currentTimeMillis()), family, qualifier,
      System.currentTimeMillis(), qualifier));
  }
  return edit;
}

@Override
public void postWALWrite(ObserverContext<? extends WALCoprocessorEnvironment> ctx,
    HRegionInfo info, WALKey logKey, WALEdit logEdit) throws IOException {
  // only keep primary region's edits
  if (logKey.getTablename().equals(tableName) && info.getReplicaId() == 0) {
    entries.add(new Entry(logKey, logEdit));
  }
}

@Override
public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e, final Put put,
    final WALEdit edit, final Durability durability) throws IOException {
  byte[] attribute = put.getAttribute("visibility");
  byte[] cf = null;
  List<Cell> updatedCells = new ArrayList<Cell>();
  if (attribute != null) {
    for (List<? extends Cell> edits : put.getFamilyCellMap().values()) {
      for (Cell cell : edits) {
        KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
        if (cf == null) {
          cf = kv.getFamily();
        }
        Tag tag = new Tag(TAG_TYPE, attribute);
        List<Tag> tagList = new ArrayList<Tag>();
        tagList.add(tag);

        KeyValue newKV = new KeyValue(kv.getRow(), 0, kv.getRowLength(), kv.getFamily(), 0,
            kv.getFamilyLength(), kv.getQualifier(), 0, kv.getQualifierLength(),
            kv.getTimestamp(), KeyValue.Type.codeToType(kv.getType()), kv.getValue(), 0,
            kv.getValueLength(), tagList);
        ((List<Cell>) updatedCells).add(newKV);
      }
    }
    put.getFamilyCellMap().remove(cf);
    // Update the family map
    put.getFamilyCellMap().put(cf, updatedCells);
  }
}

@Override
public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put m, WALEdit edit,
    Durability durability) throws IOException {
  byte[] attribute = m.getAttribute(NON_VISIBILITY);
  byte[] cf = null;
  List<Cell> updatedCells = new ArrayList<Cell>();
  if (attribute != null) {
    for (List<? extends Cell> edits : m.getFamilyCellMap().values()) {
      for (Cell cell : edits) {
        KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
        if (cf == null) {
          cf = kv.getFamily();
        }
        Tag tag = new Tag((byte) NON_VIS_TAG_TYPE, attribute);
        List<Tag> tagList = new ArrayList<Tag>();
        tagList.add(tag);
        tagList.addAll(kv.getTags());
        byte[] fromList = Tag.fromList(tagList);
        TagRewriteCell newcell = new TagRewriteCell(kv, fromList);
        KeyValue newKV = new KeyValue(kv.getRow(), 0, kv.getRowLength(), kv.getFamily(), 0,
            kv.getFamilyLength(), kv.getQualifier(), 0, kv.getQualifierLength(),
            kv.getTimestamp(), KeyValue.Type.codeToType(kv.getType()), kv.getValue(), 0,
            kv.getValueLength(), tagList);
        ((List<Cell>) updatedCells).add(newcell);
      }
    }
    m.getFamilyCellMap().remove(cf);
    // Update the family map
    m.getFamilyCellMap().put(cf, updatedCells);
  }
}

@Override
public long append(HTableDescriptor htd, HRegionInfo info, WALKey key, WALEdit edits,
    boolean inMemstore) throws IOException {
  if (this.ft == FailureType.APPEND) {
    throw new IOException("append");
  }
  return super.append(htd, info, key, edits, inMemstore);
}

@Override
public void postPut(final ObserverContext<RegionCoprocessorEnvironment> c,
    final Put put, final WALEdit edit,
    final Durability durability)
    throws IOException {
  id = System.currentTimeMillis();
  try {
    Thread.sleep(10);
  } catch (InterruptedException ex) {
  }
}

@Override
public void setup(Mapper<WALKey, WALEdit, ImmutableBytesWritable, Mutation>.Context context)
    throws IOException {
  super.setup(context);
  try {
    this.keysToFind = readKeysToSearch(context.getConfiguration());
    LOG.info("Loaded keys to find: count=" + this.keysToFind.size());
  } catch (InterruptedException e) {
    throw new InterruptedIOException(e.toString());
  }
}

private void addFamilyMapToWALEdit(Map<byte[], List<Cell>> familyMap,
    WALEdit walEdit) {
  for (List<Cell> edits : familyMap.values()) {
    for (Cell cell : edits) {
      walEdit.add(cell);
    }
  }
}

private Path writeWAL(final WALFactory wals, final String tblName) throws IOException {
  Configuration conf = TEST_UTIL.getConfiguration();
  String clsName = conf.get(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, WALCellCodec.class.getName());
  conf.setClass(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, SecureWALCellCodec.class,
    WALCellCodec.class);
  try {
    TableName tableName = TableName.valueOf(tblName);
    HTableDescriptor htd = new HTableDescriptor(tableName);
    htd.addFamily(new HColumnDescriptor(tableName.getName()));
    HRegionInfo regioninfo = new HRegionInfo(tableName,
      HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, false);
    final int total = 10;
    final byte[] row = Bytes.toBytes("row");
    final byte[] family = Bytes.toBytes("family");
    final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);

    // Write the WAL
    WAL wal = wals.getWAL(regioninfo.getEncodedNameAsBytes());
    for (int i = 0; i < total; i++) {
      WALEdit kvs = new WALEdit();
      kvs.add(new KeyValue(row, family, Bytes.toBytes(i), value));
      wal.append(htd, regioninfo, new WALKey(regioninfo.getEncodedNameAsBytes(), tableName,
          System.currentTimeMillis(), mvcc), kvs, true);
    }
    wal.sync();
    final Path walPath = DefaultWALProvider.getCurrentFileName(wal);
    wal.shutdown();

    return walPath;
  } finally {
    // restore the cell codec class
    conf.set(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, clsName);
  }
}

protected void addEdits(WAL log, HRegionInfo hri, HTableDescriptor htd,
                      int times) 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));
    log.append(htd, hri, getWalKey(hri.getEncodedNameAsBytes(), htd.getTableName(), timestamp),
      cols, true);
  }
  log.sync();
}