@Test
  public void testScanWithBatchSizeReturnIncompleteCells()
    throws IOException, InterruptedException {
    TableDescriptor hd = TableDescriptorBuilder.newBuilder(tableName)
      .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(FAMILY).setMaxVersions(3).build())
      .build();
    try (Table table = TEST_UTIL.createTable(hd, null)) {
      TEST_UTIL.waitTableAvailable(tableName, WAITTABLE_MILLIS);

      Put put = new Put(ROW);
      put.addColumn(FAMILY, Bytes.toBytes(0), generateHugeValue(3 * 1024 * 1024));
      table.put(put);

      put = new Put(ROW);
      put.addColumn(FAMILY, Bytes.toBytes(1), generateHugeValue(4 * 1024 * 1024));
      table.put(put);

      for (int i = 2; i < 5; i++) {
        for (int version = 0; version < 2; version++) {
          put = new Put(ROW);
          put.addColumn(FAMILY, Bytes.toBytes(i), generateHugeValue(1024));
          table.put(put);
        }
      }

      Scan scan = new Scan();
      scan.withStartRow(ROW).withStopRow(ROW, true).addFamily(FAMILY).setBatch(3)
        .setMaxResultSize(4 * 1024 * 1024);
      Result result;
      try (ResultScanner scanner = table.getScanner(scan)) {
        List<Result> list = new ArrayList<>();
        /*
         * The first scan rpc should return a result with 2 cells, because 3MB + 4MB > 4MB; The
         * second scan rpc should return a result with 3 cells, because reach the batch limit = 3;
         * The mayHaveMoreCellsInRow in last result should be false in the scan rpc. BTW, the
         * moreResultsInRegion also would be false. Finally, the client should collect all the cells
         * into two result: 2+3 -> 3+2;
         */
        while ((result = scanner.next()) != null) {
          list.add(result);
        }

        Assert.assertEquals(5, list.stream().mapToInt(Result::size).sum());
        Assert.assertEquals(2, list.size());
        Assert.assertEquals(3, list.get(0).size());
        Assert.assertEquals(2, list.get(1).size());
      }

      scan = new Scan();
      scan.withStartRow(ROW).withStopRow(ROW, true).addFamily(FAMILY).setBatch(2)
        .setMaxResultSize(4 * 1024 * 1024);
      try (ResultScanner scanner = table.getScanner(scan)) {
        List<Result> list = new ArrayList<>();
        while ((result = scanner.next()) != null) {
          list.add(result);
        }
        Assert.assertEquals(5, list.stream().mapToInt(Result::size).sum());
        Assert.assertEquals(3, list.size());
        Assert.assertEquals(2, list.get(0).size());
        Assert.assertEquals(2, list.get(1).size());
        Assert.assertEquals(1, list.get(2).size());
      }
    }
  }
A test case for issue HBASE-17482 After combile seqid with mvcc readpoint, seqid/mvcc is
acquired and stamped onto cells in the append thread, a countdown latch is used to ensure that
happened before cells can be put into memstore. But the MVCCPreAssign patch(HBASE-16698) make
the seqid/mvcc acquirement in handler thread and stamping in append thread No countdown latch
to assure cells in memstore are stamped with seqid/mvcc. If cells without mvcc(A.K.A mvcc=0)
are put into memstore, then a scanner with a smaller readpoint can see these data, which
disobey the multi version concurrency control rules. This test case is to reproduce this
scenario.
hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide3.java