HBASE-17510 DefaultMemStore gets the wrong heap size after rollback (ChiaPing Tsai)

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

@@ -347,8 +347,8 @@ public class DefaultMemStore implements MemStore {
     // If the key is in the memstore, delete it. Update this.size.
     found = this.cellSet.get(cell);
     if (found != null && found.getSequenceId() == cell.getSequenceId()) {
-      removeFromCellSet(cell);
+      removeFromCellSet(found);
-      long s = heapSizeChange(cell, true);
+      long s = heapSizeChange(found, true);
       this.size.addAndGet(-s);
     }
   }

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

@@ -90,6 +90,7 @@ import org.junit.rules.TestName;
 import org.mockito.Mockito;
 
 import com.google.common.collect.Lists;
+import java.util.Arrays;
 
 /**
  * Test class for the Store
@@ -358,6 +359,29 @@ public class TestStore {
     }
   }
 
+  @Test
+  public void testRollback() throws IOException {
+    Configuration conf = HBaseConfiguration.create();
+    FileSystem fs = FileSystem.get(conf);
+    // Initialize region
+    init(name.getMethodName(), conf);
+    Cell cell = CellUtil.createCell(row, family, qf1);
+    int len = KeyValueUtil.length(cell);
+    int offset = 77;
+    byte[] buf = new byte[offset + len];
+    KeyValueUtil.appendToByteArray(cell, buf, offset);
+    KeyValue newKv = new KeyValue(buf, offset, len);
+    newKv.setSequenceId(cell.getSequenceId());
+    List<Cell> testCells = Arrays.asList(cell, cell, newKv);
+    for (Cell c : testCells) {
+      long sizeBeforeRollback = store.heapSize();
+      store.add(cell);
+      store.rollback(cell);
+      long sizeAeforeRollback = store.heapSize();
+      assertEquals(sizeBeforeRollback, sizeAeforeRollback);
+    }
+  }
+
   @Test
   public void testLowestModificationTime() throws Exception {
     Configuration conf = HBaseConfiguration.create();