HBASE-17885 Backport HBASE-15871 to branch-1

HBASE-15871 Memstore flush doesn't finish because of backwardseek() in memstore scanner (ramkrishna.s.vasudevan)

Signed-off-by: Andrew Purtell <apurtell@apache.org>

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

@@ -647,8 +647,14 @@ public class DefaultMemStore implements MemStore {
    */
   @Override
   public List<KeyValueScanner> getScanners(long readPt) {
-    return Collections.<KeyValueScanner> singletonList(
-        new MemStoreScanner(activeSection, snapshotSection, readPt, comparator));
+    MemStoreScanner scanner =
+      new MemStoreScanner(activeSection, snapshotSection, readPt, comparator);
+    scanner.seek(CellUtil.createCell(HConstants.EMPTY_START_ROW));
+    if (scanner.peek() == null) {
+      scanner.close();
+      return null;
+    }
+    return Collections.<KeyValueScanner> singletonList(scanner);
   }
 
   /**

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

@@ -1220,7 +1220,9 @@ public class HStore implements Store {
       new ArrayList<KeyValueScanner>(sfScanners.size()+1);
     scanners.addAll(sfScanners);
     // Then the memstore scanners
-    scanners.addAll(memStoreScanners);
+    if (memStoreScanners != null) {
+      scanners.addAll(memStoreScanners);
+    }
     return scanners;
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java

@@ -177,7 +177,7 @@ public class TestLoadIncrementalHFiles {
    * Test case that creates some regions and loads HFiles that cross the boundaries
    * and have different region boundaries than the table pre-split.
    */
-  @Test(timeout = 60000)
+  @Test(timeout = 80000)
   public void testRegionCrossingHFileSplit() throws Exception {
     testRegionCrossingHFileSplit(BloomType.NONE);
   }

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

@@ -276,12 +276,11 @@ public class TestDefaultMemStore extends TestCase {
     kv1.setSequenceId(w.getWriteNumber());
     memstore.add(kv1);
 
-    KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
-    assertScannerResults(s, new KeyValue[]{});
+    assertTrue(this.memstore.getScanners(mvcc.getReadPoint()) == null);
 
     mvcc.completeAndWait(w);
 
-    s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
+    KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
     assertScannerResults(s, new KeyValue[]{kv1});
 
     w = mvcc.begin();

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

@@ -6809,4 +6809,65 @@ public class TestHRegion {
       this.region = null;
     }
   }
+
+  @Test
+  public void testReverseScanShouldNotScanMemstoreIfReadPtLesser() throws Exception {
+    byte[] cf1 = Bytes.toBytes("CF1");
+    byte[][] families = { cf1 };
+    byte[] col = Bytes.toBytes("C");
+    String method = this.getName();
+    HBaseConfiguration conf = new HBaseConfiguration();
+    this.region = initHRegion(tableName, method, conf, families);
+    try {
+      // setup with one storefile and one memstore, to create scanner and get an earlier readPt
+      Put put = new Put(Bytes.toBytes("19996"));
+      put.addColumn(cf1, col, Bytes.toBytes("val"));
+      region.put(put);
+      Put put2 = new Put(Bytes.toBytes("19995"));
+      put2.addColumn(cf1, col, Bytes.toBytes("val"));
+      region.put(put2);
+
+      // create a reverse scan
+      Scan scan = new Scan(Bytes.toBytes("19996"));
+      scan.setReversed(true);
+      RegionScanner scanner = region.getScanner(scan);
+
+      // flush the cache. This will reset the store scanner
+      region.flushcache(true, true);
+
+      // create one memstore contains many rows will be skipped
+      // to check MemStoreScanner.seekToPreviousRow
+      for (int i = 10000; i < 20000; i++) {
+        Put p = new Put(Bytes.toBytes("" + i));
+        p.addColumn(cf1, col, Bytes.toBytes("" + i));
+        region.put(p);
+      }
+
+      List<Cell> currRow = new ArrayList<>();
+      boolean hasNext;
+      boolean assertDone = false;
+      do {
+        hasNext = scanner.next(currRow);
+        // With HBASE-15871, after the scanner is reset the memstore scanner should not be
+        // added here
+        if (!assertDone) {
+          StoreScanner current =
+            (StoreScanner) (((RegionScannerImpl) scanner).storeHeap).getCurrentForTesting();
+          List<KeyValueScanner> scanners = current.getAllScannersForTesting();
+          assertEquals("There should be only one scanner the store file scanner", 1,
+            scanners.size());
+          assertDone = true;
+        }
+      } while (hasNext);
+
+      assertEquals(2, currRow.size());
+      assertEquals("19996", Bytes.toString(currRow.get(0).getRowArray(),
+        currRow.get(0).getRowOffset(), currRow.get(0).getRowLength()));
+      assertEquals("19995", Bytes.toString(currRow.get(1).getRowArray(),
+        currRow.get(1).getRowOffset(), currRow.get(1).getRowLength()));
+    } finally {
+      HBaseTestingUtility.closeRegionAndWAL(this.region);
+      this.region = null;
+    }
+  }
 }