HBASE-17662 Disable in-memory flush when replaying from WAL

Signed-off-by: anoopsamjohn <anoopsamjohn@gmail.com>

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

@@ -54,8 +54,9 @@ public abstract class AbstractMemStore implements MemStore {
   // Used to track when to flush
   private volatile long timeOfOldestEdit;
 
-  public final static long FIXED_OVERHEAD = ClassSize
+  public final static long FIXED_OVERHEAD = ClassSize.OBJECT
-      .align(ClassSize.OBJECT + (4 * ClassSize.REFERENCE) + (2 * Bytes.SIZEOF_LONG));
+          + (4 * ClassSize.REFERENCE)
+          + (2 * Bytes.SIZEOF_LONG); // snapshotId, timeOfOldestEdit
 
   public final static long DEEP_OVERHEAD = FIXED_OVERHEAD;
 

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

@@ -71,16 +71,22 @@ public class CompactingMemStore extends AbstractMemStore {
 
   private long inmemoryFlushSize;       // the threshold on active size for in-memory flush
   private final AtomicBoolean inMemoryFlushInProgress = new AtomicBoolean(false);
+
+  // inWalReplay is true while we are synchronously replaying the edits from WAL
+  private boolean inWalReplay = false;
+
   @VisibleForTesting
   private final AtomicBoolean allowCompaction = new AtomicBoolean(true);
   private boolean compositeSnapshot = true;
 
-  public static final long DEEP_OVERHEAD = AbstractMemStore.DEEP_OVERHEAD
+
-      + 6 * ClassSize.REFERENCE // Store, RegionServicesForStores, CompactionPipeline,
+  public static final long DEEP_OVERHEAD = ClassSize.align( AbstractMemStore.DEEP_OVERHEAD
-                                // MemStoreCompactor, inMemoryFlushInProgress, allowCompaction
+      + 6 * ClassSize.REFERENCE     // Store, RegionServicesForStores, CompactionPipeline,
-      + Bytes.SIZEOF_LONG // inmemoryFlushSize
+                                    // MemStoreCompactor, inMemoryFlushInProgress, allowCompaction
+      + Bytes.SIZEOF_LONG           // inmemoryFlushSize
+      + 2 * Bytes.SIZEOF_BOOLEAN    // compositeSnapshot and inWalReplay
       + 2 * ClassSize.ATOMIC_BOOLEAN// inMemoryFlushInProgress and allowCompaction
-      + CompactionPipeline.DEEP_OVERHEAD + MemStoreCompactor.DEEP_OVERHEAD;
+      + CompactionPipeline.DEEP_OVERHEAD + MemStoreCompactor.DEEP_OVERHEAD);
 
   public CompactingMemStore(Configuration conf, CellComparator c,
       HStore store, RegionServicesForStores regionServices,
@@ -232,6 +238,24 @@ public class CompactingMemStore extends AbstractMemStore {
     }
   }
 
+  /**
+   * This message intends to inform the MemStore that next coming updates
+   * are going to be part of the replaying edits from WAL
+   */
+  @Override
+  public void startReplayingFromWAL() {
+    inWalReplay = true;
+  }
+
+  /**
+   * This message intends to inform the MemStore that the replaying edits from WAL
+   * are done
+   */
+  @Override
+  public void stopReplayingFromWAL() {
+    inWalReplay = false;
+  }
+
   // the getSegments() method is used for tests only
   @VisibleForTesting
   @Override
@@ -388,9 +412,12 @@ public class CompactingMemStore extends AbstractMemStore {
 
   private boolean shouldFlushInMemory() {
     if (this.active.keySize() > inmemoryFlushSize) { // size above flush threshold
-        // the inMemoryFlushInProgress is CASed to be true here in order to mutual exclude
+      if (inWalReplay) {  // when replaying edits from WAL there is no need in in-memory flush
-        // the insert of the active into the compaction pipeline
+        return false;     // regardless the size
-        return (inMemoryFlushInProgress.compareAndSet(false,true));
+      }
+      // the inMemoryFlushInProgress is CASed to be true here in order to mutual exclude
+      // the insert of the active into the compaction pipeline
+      return (inMemoryFlushInProgress.compareAndSet(false,true));
     }
     return false;
   }

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

@@ -35,6 +35,7 @@ import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.ClassSize;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 
 /**
@@ -57,6 +58,8 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 public class DefaultMemStore extends AbstractMemStore {
   private static final Log LOG = LogFactory.getLog(DefaultMemStore.class);
 
+  public final static long DEEP_OVERHEAD = ClassSize.align(AbstractMemStore.DEEP_OVERHEAD);
+  public final static long FIXED_OVERHEAD = ClassSize.align(AbstractMemStore.FIXED_OVERHEAD);
   /**
    * Default constructor. Used for tests.
    */

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

@@ -878,11 +878,22 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
     long maxSeqId = initializeStores(reporter, status);
     this.mvcc.advanceTo(maxSeqId);
     if (ServerRegionReplicaUtil.shouldReplayRecoveredEdits(this)) {
-      // Recover any edits if available.
+      List<Store> stores = this.getStores();  // update the stores that we are replaying
-      maxSeqId = Math.max(maxSeqId,
+      try {
-        replayRecoveredEditsIfAny(this.fs.getRegionDir(), maxSeqIdInStores, reporter, status));
+        for (Store store : stores) {
-      // Make sure mvcc is up to max.
+          ((HStore) store).startReplayingFromWAL();
-      this.mvcc.advanceTo(maxSeqId);
+        }
+        // Recover any edits if available.
+        maxSeqId = Math.max(maxSeqId,
+            replayRecoveredEditsIfAny(this.fs.getRegionDir(), maxSeqIdInStores, reporter, status));
+        // Make sure mvcc is up to max.
+        this.mvcc.advanceTo(maxSeqId);
+      } finally {
+        for (Store store : stores) {            // update the stores that we are done replaying
+          ((HStore)store).stopReplayingFromWAL();
+        }
+      }
+
     }
     this.lastReplayedOpenRegionSeqId = maxSeqId;
 

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

@@ -662,6 +662,22 @@ public class HStore implements Store {
     return storeFile;
   }
 
+  /**
+   * This message intends to inform the MemStore that next coming updates
+   * are going to be part of the replaying edits from WAL
+   */
+  public void startReplayingFromWAL(){
+    this.memstore.startReplayingFromWAL();
+  }
+
+  /**
+   * This message intends to inform the MemStore that the replaying edits from WAL
+   * are done
+   */
+  public void stopReplayingFromWAL(){
+    this.memstore.stopReplayingFromWAL();
+  }
+
   /**
    * Adds a value to the memstore
    * @param cell

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

@@ -127,4 +127,16 @@ public interface MemStore {
 
   /* Return true if the memstore may use some extra memory space*/
   boolean isSloppy();
+
+  /**
+   * This message intends to inform the MemStore that next coming updates
+   * are going to be part of the replaying edits from WAL
+   */
+  default void startReplayingFromWAL(){return;}
+
+  /**
+   * This message intends to inform the MemStore that the replaying edits from WAL
+   * are done
+   */
+  default void stopReplayingFromWAL(){return;}
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java

@@ -324,10 +324,10 @@ public class TestHeapSize  {
     expected += ClassSize.estimateBase(AtomicBoolean.class, false);
     expected += ClassSize.estimateBase(AtomicBoolean.class, false);
     expected += ClassSize.estimateBase(CompactionPipeline.class, false);
-    expected += ClassSize.estimateBase(LinkedList.class, false);
+    expected += ClassSize.estimateBase(LinkedList.class, false); //inside CompactionPipeline
-    expected += ClassSize.estimateBase(LinkedList.class, false);
+    expected += ClassSize.estimateBase(LinkedList.class, false); //inside CompactionPipeline
     expected += ClassSize.estimateBase(MemStoreCompactor.class, false);
-    expected += ClassSize.estimateBase(AtomicBoolean.class, false);
+    expected += ClassSize.estimateBase(AtomicBoolean.class, false);// inside MemStoreCompactor
     if (expected != actual) {
       ClassSize.estimateBase(cl, true);
       ClassSize.estimateBase(AtomicBoolean.class, true);

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/AbstractTestWALReplay.java

@@ -127,8 +127,6 @@ public abstract class AbstractTestWALReplay {
     Configuration conf = TEST_UTIL.getConfiguration();
     // The below config supported by 0.20-append and CDH3b2
     conf.setInt("dfs.client.block.recovery.retries", 2);
-    conf.set(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY,
-        String.valueOf(MemoryCompactionPolicy.NONE));
     TEST_UTIL.startMiniCluster(3);
     Path hbaseRootDir =
       TEST_UTIL.getDFSCluster().getFileSystem().makeQualified(new Path("/hbase"));