HBASE-23221 Polish the WAL interface after HBASE-23181 (#774)

Removes the closeRegion flag added by HBASE-23181 and instead
relies on reading meta WALEdit content. Modified how qualifier is
written when the meta WALEdit is for a RegionEventDescriptor
so the 'type' is added to the qualifer so can figure type
w/o having to deserialize protobuf value content: e.g.
HBASE::REGION_EVENT::REGION_CLOSE

Added doc on WALEdit and tried to formalize the 'meta' WALEdit
type and how it works. Needs complete redo in part as suggested
by HBASE-8457. Meantime, some doc and cleanup.

Also changed the LogRoller constructor to remove redundant param.
Because of constructor change, need to change also
TestFailedAppendAndSync, TestWALLockup, TestAsyncFSWAL &
WALPerformanceEvaluation.java

Signed-off-by: Duo Zhang <zhangduo@apache.org>
Signed-off-by: Lijin Bin <binlijin@apache.org>

Author: saintstack

hbase-common/src/main/java/org/apache/hadoop/hbase/CellUtil.java

@@ -756,7 +756,7 @@ public static boolean matchingRow(final Cell left, final Cell right) {
 
   /**
    * @deprecated As of release 2.0.0, this will be removed in HBase 3.0.0. Instead use
-   *             {@link #matchingRows(Cell, byte[]))}
+   *   {@link #matchingRows(Cell, byte[])}
    */
   @Deprecated
   public static boolean matchingRow(final Cell left, final byte[] buf) {
@@ -894,8 +894,15 @@ public static boolean matchingQualifier(final Cell left, final byte[] buf, final
   }
 
   public static boolean matchingColumn(final Cell left, final byte[] fam, final byte[] qual) {
-    if (!matchingFamily(left, fam)) return false;
-    return matchingQualifier(left, qual);
+    return matchingFamily(left, fam) && matchingQualifier(left, qual);
+  }
+
+  /**
+   * @return True if matching column family and the qualifier starts with <code>qual</code>
+   */
+  public static boolean matchingColumnFamilyAndQualifierPrefix(final Cell left, final byte[] fam,
+      final byte[] qual) {
+    return matchingFamily(left, fam) && PrivateCellUtil.qualifierStartsWith(left, qual);
   }
 
   /**

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALPlayer.java

@@ -199,11 +199,10 @@ public void map(WALKey key, WALEdit value, Context context)
           Delete del = null;
           Cell lastCell = null;
           for (Cell cell : value.getCells()) {
-            // filtering WAL meta entries
+            // Filtering WAL meta marker entries.
             if (WALEdit.isMetaEditFamily(cell)) {
               continue;
             }
-
             // Allow a subclass filter out this cell.
             if (filter(context, cell)) {
               // A WALEdit may contain multiple operations (HBASE-3584) and/or

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -3566,7 +3566,7 @@ public List<ReplicationPeerDescription> listReplicationPeers(String regex)
     if (cpHost != null) {
       cpHost.preListReplicationPeers(regex);
     }
-    LOG.info(getClientIdAuditPrefix() + " list replication peers, regex=" + regex);
+    LOG.debug("{} list replication peers, regex={}", getClientIdAuditPrefix(), regex);
     Pattern pattern = regex == null ? null : Pattern.compile(regex);
     List<ReplicationPeerDescription> peers =
       this.replicationPeerManager.listPeers(pattern);

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -1725,7 +1725,7 @@ public Pair<byte[], Collection<HStoreFile>> call() throws IOException {
 
       status.setStatus("Writing region close event to WAL");
       // Always write close marker to wal even for read only table. This is not a big problem as we
-      // do not write any data into the region.
+      // do not write any data into the region; it is just a meta edit in the WAL file.
       if (!abort && wal != null && getRegionServerServices() != null &&
         RegionReplicaUtil.isDefaultReplica(getRegionInfo())) {
         writeRegionCloseMarker(wal);
@@ -2691,7 +2691,8 @@ private void logFatLineOnFlush(Collection<HStore> storesToFlush, long sequenceId
       }
     }
     MemStoreSize mss = this.memStoreSizing.getMemStoreSize();
-    LOG.info("Flushing " + storesToFlush.size() + "/" + stores.size() + " column families," +
+    LOG.info("Flushing " + this.getRegionInfo().getEncodedName() + " " +
+        storesToFlush.size() + "/" + stores.size() + " column families," +
         " dataSize=" + StringUtils.byteDesc(mss.getDataSize()) +
         " heapSize=" + StringUtils.byteDesc(mss.getHeapSize()) +
         ((perCfExtras != null && perCfExtras.length() > 0)? perCfExtras.toString(): "") +
@@ -4817,7 +4818,7 @@ private long replayRecoveredEdits(final Path edits,
           for (Cell cell: val.getCells()) {
             // Check this edit is for me. Also, guard against writing the special
             // METACOLUMN info such as HBASE::CACHEFLUSH entries
-            if (CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
+            if (WALEdit.isMetaEditFamily(cell)) {
               // if region names don't match, skipp replaying compaction marker
               if (!checkRowWithinBoundary) {
                 //this is a special edit, we should handle it

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -1926,7 +1926,7 @@ private void startServices() throws IOException {
       healthCheckChore = new HealthCheckChore(sleepTime, this, getConfiguration());
     }
 
-    this.walRoller = new LogRoller(this, this);
+    this.walRoller = new LogRoller(this);
     this.flushThroughputController = FlushThroughputControllerFactory.create(this, conf);
     this.procedureResultReporter = new RemoteProcedureResultReporter(this);
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java

@@ -1,4 +1,4 @@
-/**
+/*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -28,7 +28,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.Server;
 import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
@@ -56,7 +55,6 @@
 public class LogRoller extends HasThread implements Closeable {
   private static final Logger LOG = LoggerFactory.getLogger(LogRoller.class);
   private final ConcurrentMap<WAL, Boolean> walNeedsRoll = new ConcurrentHashMap<>();
-  private final Server server;
   protected final RegionServerServices services;
   private volatile long lastRollTime = System.currentTimeMillis();
   // Period to roll log.
@@ -99,16 +97,14 @@ public void requestRollAll() {
     }
   }
 
-  /** @param server */
-  public LogRoller(final Server server, final RegionServerServices services) {
+  public LogRoller(RegionServerServices services) {
     super("LogRoller");
-    this.server = server;
     this.services = services;
-    this.rollPeriod = this.server.getConfiguration().
+    this.rollPeriod = this.services.getConfiguration().
       getLong("hbase.regionserver.logroll.period", 3600000);
-    this.threadWakeFrequency = this.server.getConfiguration().
+    this.threadWakeFrequency = this.services.getConfiguration().
       getInt(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
-    this.checkLowReplicationInterval = this.server.getConfiguration().getLong(
+    this.checkLowReplicationInterval = this.services.getConfiguration().getLong(
         "hbase.regionserver.hlog.check.lowreplication.interval", 30 * 1000);
   }
 
@@ -144,7 +140,7 @@ private void abort(String reason, Throwable cause) {
         LOG.warn("Failed to shutdown wal", e);
       }
     }
-    server.abort(reason, cause);
+    this.services.abort(reason, cause);
   }
 
   @Override
@@ -156,7 +152,7 @@ public void run() {
       periodic = (now - this.lastRollTime) > this.rollPeriod;
       if (periodic) {
         // Time for periodic roll, fall through
-        LOG.debug("Wal roll period {} ms elapsed", this.rollPeriod);
+        LOG.debug("WAL roll period {} ms elapsed", this.rollPeriod);
       } else {
         synchronized (this) {
           if (walNeedsRoll.values().stream().anyMatch(Boolean::booleanValue)) {
@@ -183,9 +179,9 @@ public void run() {
           WAL wal = entry.getKey();
           // reset the flag in front to avoid missing roll request before we return from rollWriter.
           walNeedsRoll.put(wal, Boolean.FALSE);
-            // Force the roll if the logroll.period is elapsed or if a roll was requested.
-            // The returned value is an array of actual region names.
-            byte[][]   regionsToFlush = wal.rollWriter(periodic || entry.getValue().booleanValue());
+          // Force the roll if the logroll.period is elapsed or if a roll was requested.
+          // The returned value is an array of actual region names.
+          byte[][] regionsToFlush = wal.rollWriter(periodic || entry.getValue().booleanValue());
           if (regionsToFlush != null) {
             for (byte[] r : regionsToFlush) {
               scheduleFlush(Bytes.toString(r));

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java

@@ -261,8 +261,8 @@ public abstract class AbstractFSWAL<W extends WriterBase> implements WAL {
   private static final class WalProps {
 
     /**
-     * Map the encoded region name to the highest sequence id. Contain all the regions it has
-     * entries of
+     * Map the encoded region name to the highest sequence id.
+     * <p/>Contains all the regions it has an entry for.
      */
     public final Map<byte[], Long> encodedName2HighestSequenceId;
 
@@ -610,9 +610,9 @@ public int getNumLogFiles() {
   }
 
   /**
-   * If the number of un-archived WAL files is greater than maximum allowed, check the first
-   * (oldest) WAL file, and returns those regions which should be flushed so that it can be
-   * archived.
+   * If the number of un-archived WAL files ('live' WALs) is greater than maximum allowed,
+   * check the first (oldest) WAL, and return those regions which should be flushed so that
+   * it can be let-go/'archived'.
    * @return regions (encodedRegionNames) to flush in order to archive oldest WAL file.
    */
   byte[][] findRegionsToForceFlush() throws IOException {
@@ -888,10 +888,6 @@ public void close() throws IOException {
   /**
    * updates the sequence number of a specific store. depending on the flag: replaces current seq
    * number if the given seq id is bigger, or even if it is lower than existing one
-   * @param encodedRegionName
-   * @param familyName
-   * @param sequenceid
-   * @param onlyIfGreater
    */
   @Override
   public void updateStore(byte[] encodedRegionName, byte[] familyName, Long sequenceid,
@@ -1015,7 +1011,7 @@ protected final void postSync(final long timeInNanos, final int handlerSyncs) {
   }
 
   protected final long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKeyImpl key,
-    WALEdit edits, boolean inMemstore, boolean closeRegion, RingBuffer<RingBufferTruck> ringBuffer)
+    WALEdit edits, boolean inMemstore, RingBuffer<RingBufferTruck> ringBuffer)
     throws IOException {
     if (this.closed) {
       throw new IOException(
@@ -1029,7 +1025,7 @@ protected final long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKe
     ServerCall<?> rpcCall = RpcServer.getCurrentCall().filter(c -> c instanceof ServerCall)
       .filter(c -> c.getCellScanner() != null).map(c -> (ServerCall) c).orElse(null);
     try (TraceScope scope = TraceUtil.createTrace(implClassName + ".append")) {
-      FSWALEntry entry = new FSWALEntry(txid, key, edits, hri, inMemstore, closeRegion, rpcCall);
+      FSWALEntry entry = new FSWALEntry(txid, key, edits, hri, inMemstore, rpcCall);
       entry.stampRegionSequenceId(we);
       ringBuffer.get(txid).load(entry);
     } finally {
@@ -1067,13 +1063,13 @@ public OptionalLong getLogFileSizeIfBeingWritten(Path path) {
 
   @Override
   public long appendData(RegionInfo info, WALKeyImpl key, WALEdit edits) throws IOException {
-    return append(info, key, edits, true, false);
+    return append(info, key, edits, true);
   }
 
   @Override
-  public long appendMarker(RegionInfo info, WALKeyImpl key, WALEdit edits, boolean closeRegion)
+  public long appendMarker(RegionInfo info, WALKeyImpl key, WALEdit edits)
     throws IOException {
-    return append(info, key, edits, false, closeRegion);
+    return append(info, key, edits, false);
   }
 
   /**
@@ -1097,17 +1093,17 @@ public long appendMarker(RegionInfo info, WALKeyImpl key, WALEdit edits, boolean
    * @param key Modified by this call; we add to it this edits region edit/sequence id.
    * @param edits Edits to append. MAY CONTAIN NO EDITS for case where we want to get an edit
    *          sequence id that is after all currently appended edits.
-   * @param inMemstore Always true except for case where we are writing a region event marker, for
-   *          example, a compaction completion record into the WAL; in this case the entry is just
-   *          so we can finish an unfinished compaction -- it is not an edit for memstore.
-   * @param closeRegion Whether this is a region close marker, i.e, the last wal edit for this
-   *          region on this region server. The WAL implementation should remove all the related
-   *          stuff, for example, the sequence id accounting.
+   * @param inMemstore Always true except for case where we are writing a region event meta
+   *          marker edit, for example, a compaction completion record into the WAL or noting a
+   *          Region Open event. In these cases the entry is just so we can finish an unfinished
+   *          compaction after a crash when the new Server reads the WAL on recovery, etc. These
+   *          transition event 'Markers' do not go via the memstore. When memstore is false,
+   *          we presume a Marker event edit.
    * @return Returns a 'transaction id' and <code>key</code> will have the region edit/sequence id
    *         in it.
    */
-  protected abstract long append(RegionInfo info, WALKeyImpl key, WALEdit edits, boolean inMemstore,
-    boolean closeRegion) throws IOException;
+  protected abstract long append(RegionInfo info, WALKeyImpl key, WALEdit edits, boolean inMemstore)
+      throws IOException;
 
   protected abstract void doAppend(W writer, FSWALEntry entry) throws IOException;
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java

@@ -564,9 +564,9 @@ private boolean shouldScheduleConsumer() {
   }
 
   @Override
-  protected long append(RegionInfo hri, WALKeyImpl key, WALEdit edits, boolean inMemstore,
-    boolean closeRegion) throws IOException {
-    long txid = stampSequenceIdAndPublishToRingBuffer(hri, key, edits, inMemstore, closeRegion,
+  protected long append(RegionInfo hri, WALKeyImpl key, WALEdit edits, boolean inMemstore)
+      throws IOException {
+    long txid = stampSequenceIdAndPublishToRingBuffer(hri, key, edits, inMemstore,
       waitingConsumePayloads);
     if (shouldScheduleConsumer()) {
       consumeExecutor.execute(consumer);

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java

@@ -436,8 +436,8 @@ protected void doShutdown() throws IOException {
 
   @Override
   protected long append(final RegionInfo hri, final WALKeyImpl key, final WALEdit edits,
-    final boolean inMemstore, boolean closeRegion) throws IOException {
-    return stampSequenceIdAndPublishToRingBuffer(hri, key, edits, inMemstore, closeRegion,
+    final boolean inMemstore) throws IOException {
+    return stampSequenceIdAndPublishToRingBuffer(hri, key, edits, inMemstore,
       disruptor.getRingBuffer());
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSWALEntry.java

@@ -40,7 +40,7 @@
 /**
  * A WAL Entry for {@link AbstractFSWAL} implementation.  Immutable.
  * A subclass of {@link Entry} that carries extra info across the ring buffer such as
- * region sequence id (we want to use this later, just before we write the WAL to ensure region
+ * region sequenceid (we want to use this later, just before we write the WAL to ensure region
  * edits maintain order).  The extra info added here is not 'serialized' as part of the WALEdit
  * hence marked 'transient' to underline this fact.  It also adds mechanism so we can wait on
  * the assign of the region sequence id.  See #stampRegionSequenceId().
@@ -50,25 +50,40 @@ class FSWALEntry extends Entry {
   // The below data members are denoted 'transient' just to highlight these are not persisted;
   // they are only in memory and held here while passing over the ring buffer.
   private final transient long txid;
+
+  /**
+   * If false, means this is a meta edit written by the hbase system itself. It was not in
+   * memstore. HBase uses these edit types to note in the log operational transitions such
+   * as compactions, flushes, or region open/closes.
+   */
   private final transient boolean inMemstore;
+
+  /**
+   * Set if this is a meta edit and it is of close region type.
+   */
   private final transient boolean closeRegion;
+
   private final transient RegionInfo regionInfo;
   private final transient Set<byte[]> familyNames;
   private final transient ServerCall<?> rpcCall;
 
+  /**
+   * @param inMemstore If true, then this is a data edit, one that came from client. If false, it
+   *   is a meta edit made by the hbase system itself and is for the WAL only.
+   */
   FSWALEntry(final long txid, final WALKeyImpl key, final WALEdit edit, final RegionInfo regionInfo,
-    final boolean inMemstore, boolean closeRegion, ServerCall<?> rpcCall) {
+    final boolean inMemstore, ServerCall<?> rpcCall) {
     super(key, edit);
     this.inMemstore = inMemstore;
-    this.closeRegion = closeRegion;
+    this.closeRegion = !inMemstore && edit.isRegionCloseMarker();
     this.regionInfo = regionInfo;
     this.txid = txid;
     if (inMemstore) {
       // construct familyNames here to reduce the work of log sinker.
       Set<byte[]> families = edit.getFamilies();
       this.familyNames = families != null ? families : collectFamilies(edit.getCells());
     } else {
-      this.familyNames = Collections.<byte[]> emptySet();
+      this.familyNames = Collections.emptySet();
     }
     this.rpcCall = rpcCall;
     if (rpcCall != null) {
@@ -83,7 +98,7 @@ static Set<byte[]> collectFamilies(List<Cell> cells) {
     } else {
       Set<byte[]> set = new TreeSet<>(Bytes.BYTES_COMPARATOR);
       for (Cell cell: cells) {
-        if (!CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
+        if (!WALEdit.isMetaEditFamily(cell)) {
           set.add(CellUtil.cloneFamily(cell));
         }
       }
@@ -94,7 +109,7 @@ static Set<byte[]> collectFamilies(List<Cell> cells) {
   @Override
   public String toString() {
     return "sequence=" + this.txid + ", " + super.toString();
-  };
+  }
 
   boolean isInMemStore() {
     return this.inMemstore;