svn commit: r1181483 - /hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

[ns...@apache.org]

Author: nspiegelberg
Date: Tue Oct 11 02:12:28 2011
New Revision: 1181483

URL: http://svn.apache.org/viewvc?rev=1181483&view=rev
Log:
HBASE-3481: max seq id in flushed file can be larger than its correct value causing data loss during recovery

Summary:
(Ryan beat me to the patch). Bringing the patch in.

For the gory details see:

https://issues.apache.org/jira/browse/HBASE-3481

Test Plan:
running unit tests right now. And will do some more kill tests as well before
commit.

DiffCamp Revision: 205376
Reviewed By: kranganathan
Reviewers: jgray, kranganathan
Commenters: dhruba
CC: dhruba, kranganathan, hbase@lists
Revert Plan:
OK

Modified:
    hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1181483&r1=1181482&r2=1181483&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Tue Oct 11 02:12:28 2011
@@ -1517,6 +1517,7 @@ public class HRegion implements HeapSize
   private long doMiniBatchPut(BatchOperationInProgress<Pair<Put, Integer>> batchOp) throws IOException {
     long now = EnvironmentEdgeManager.currentTimeMillis();
     byte[] byteNow = Bytes.toBytes(now);
+    boolean locked = false;
 
     /** Keep track of the locks we hold so we can release them in finally clause */
     List<Integer> acquiredLocks = Lists.newArrayListWithCapacity(batchOp.operations.length);
@@ -1563,6 +1564,10 @@ public class HRegion implements HeapSize
       // We've now grabbed as many puts off the list as we can
       assert numReadyToWrite > 0;
 
+
+      this.updatesLock.readLock().lock();
+      locked = true;
+
       // ------------------------------------
       // STEP 2. Update any LATEST_TIMESTAMP timestamps
       // ----------------------------------
@@ -1603,6 +1608,9 @@ public class HRegion implements HeapSize
       success = true;
       return addedSize;
     } finally {
+      if (locked)
+        this.updatesLock.readLock().unlock();
+
       for (Integer toRelease : acquiredLocks) {
         releaseRowLock(toRelease);
       }
@@ -3072,6 +3080,7 @@ public class HRegion implements HeapSize
     boolean flush = false;
     // Lock row
     Integer lid = obtainRowLock(row);
+    this.updatesLock.readLock().lock();
     long result = amount;
     try {
       Store store = stores.get(family);
@@ -3111,6 +3120,7 @@ public class HRegion implements HeapSize
       size = this.memstoreSize.addAndGet(size);
       flush = isFlushSize(size);
     } finally {
+      this.updatesLock.readLock().unlock();
       releaseRowLock(lid);
     }