HBASE-21200 Memstore flush doesn't finish because of seekToPreviousRow() in memstore scanner.

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

@@ -49,6 +49,8 @@ public class SegmentScanner implements KeyValueScanner {
   // A flag represents whether could stop skipping KeyValues for MVCC
   // if have encountered the next row. Only used for reversed scan
   private boolean stopSkippingKVsIfNextRow = false;
+  // Stop skipping KeyValues for MVCC if finish this row. Only used for reversed scan
+  private Cell stopSkippingKVsRow;
   // last iterated KVs by seek (to restore the iterator state after reseek)
   private Cell last = null;
 
@@ -203,6 +205,7 @@ public class SegmentScanner implements KeyValueScanner {
       }
       Cell firstKeyOnPreviousRow = PrivateCellUtil.createFirstOnRow(lastCellBeforeRow);
       this.stopSkippingKVsIfNextRow = true;
+      this.stopSkippingKVsRow = firstKeyOnPreviousRow;
       seek(firstKeyOnPreviousRow);
       this.stopSkippingKVsIfNextRow = false;
       if (peek() == null
@@ -344,7 +347,6 @@ public class SegmentScanner implements KeyValueScanner {
    * skipping the cells with irrelevant MVCC
    */
   protected void updateCurrent() {
-    Cell startKV = current;
     Cell next = null;
 
     try {
@@ -354,9 +356,9 @@ public class SegmentScanner implements KeyValueScanner {
           current = next;
           return;// skip irrelevant versions
         }
-        if (stopSkippingKVsIfNextRow &&   // for backwardSeek() stay in the
+        // for backwardSeek() stay in the boundaries of a single row
-            startKV != null &&        // boundaries of a single row
+        if (stopSkippingKVsIfNextRow &&
-            segment.compareRows(next, startKV) > 0) {
+            segment.compareRows(next, stopSkippingKVsRow) > 0) {
           current = null;
           return;
         }

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

@@ -5825,6 +5825,46 @@ public class TestHRegion {
     }
   }
 
+  @Test
+  public void testReverseScanWhenPutCellsAfterOpenReverseScan() throws Exception {
+    byte[] cf1 = Bytes.toBytes("CF1");
+    byte[][] families = { cf1 };
+    byte[] col = Bytes.toBytes("C");
+
+    HBaseConfiguration conf = new HBaseConfiguration();
+    this.region = initHRegion(tableName, method, conf, families);
+
+    Put put = new Put(Bytes.toBytes("199996"));
+    put.addColumn(cf1, col, Bytes.toBytes("val"));
+    region.put(put);
+    Put put2 = new Put(Bytes.toBytes("199995"));
+    put2.addColumn(cf1, col, Bytes.toBytes("val"));
+    region.put(put2);
+
+    // Create a reverse scan
+    Scan scan = new Scan(Bytes.toBytes("199996"));
+    scan.setReversed(true);
+    RegionScannerImpl scanner = region.getScanner(scan);
+
+    // Put a lot of cells that have sequenceIDs grater than the readPt of the reverse scan
+    for (int i = 100000; i < 200000; 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;
+    do {
+      hasNext = scanner.next(currRow);
+    } while (hasNext);
+
+    assertEquals(2, currRow.size());
+    assertEquals("199996", Bytes.toString(currRow.get(0).getRowArray(),
+      currRow.get(0).getRowOffset(), currRow.get(0).getRowLength()));
+    assertEquals("199995", Bytes.toString(currRow.get(1).getRowArray(),
+      currRow.get(1).getRowOffset(), currRow.get(1).getRowLength()));
+  }
+
   @Test
   public void testWriteRequestsCounter() throws IOException {
     byte[] fam = Bytes.toBytes("info");