svn commit: r1064050 - in /hbase/trunk: CHANGES.txt src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

[rawson]

Author: rawson
Date: Thu Jan 27 09:25:55 2011
New Revision: 1064050

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

Modified:
    hbase/trunk/CHANGES.txt
    hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

Modified: hbase/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hbase/trunk/CHANGES.txt?rev=1064050&r1=1064049&r2=1064050&view=diff
==============================================================================
--- hbase/trunk/CHANGES.txt (original)
+++ hbase/trunk/CHANGES.txt Thu Jan 27 09:25:55 2011
@@ -30,6 +30,8 @@ Release 0.91.0 - Unreleased
    HBASE-3456  Fix hardcoding of 20 second socket timeout down in HBaseClient
    HBASE-3476  HFile -m option need not scan key values
                (Prakash Khemani via Lars George)
+   HBASE-2481  max seq id in flushed file can be larger than its correct value 
+              causing data loss during recovery
 
 
   IMPROVEMENTS

Modified: 
hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: 
http://svn.apache.org/viewvc/hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1064050&r1=1064049&r2=1064050&view=diff
==============================================================================
--- hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java 
(original)
+++ hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java 
Thu Jan 27 09:25:55 2011
@@ -1481,6 +1481,7 @@ public class HRegion implements HeapSize
 
     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);
@@ -1547,6 +1548,10 @@ public class HRegion implements HeapSize
             byteNow);
       }
 
+
+      this.updatesLock.readLock().lock();
+      locked = true;
+
       // ------------------------------------
       // STEP 3. Write to WAL
       // ----------------------------------
@@ -1589,6 +1594,9 @@ public class HRegion implements HeapSize
       success = true;
       return addedSize;
     } finally {
+      if (locked)
+        this.updatesLock.readLock().unlock();
+
       for (Integer toRelease : acquiredLocks) {
         releaseRowLock(toRelease);
       }
@@ -3156,6 +3164,7 @@ public class HRegion implements HeapSize
     startRegionOperation();
     try {
       Integer lid = getLock(lockid, row, true);
+      this.updatesLock.readLock().lock();
       try {
         // Process each family
         for (Map.Entry<byte [], NavigableMap<byte [], Long>> family :
@@ -3211,6 +3220,7 @@ public class HRegion implements HeapSize
         size = this.memstoreSize.addAndGet(size);
         flush = isFlushSize(size);
       } finally {
+        this.updatesLock.readLock().unlock();
         releaseRowLock(lid);
       }
     } finally {
@@ -3244,6 +3254,7 @@ public class HRegion implements HeapSize
     startRegionOperation();
     try {
       Integer lid = obtainRowLock(row);
+      this.updatesLock.readLock().lock();
       try {
         Store store = stores.get(family);
 
@@ -3284,6 +3295,7 @@ public class HRegion implements HeapSize
         size = this.memstoreSize.addAndGet(size);
         flush = isFlushSize(size);
       } finally {
+        this.updatesLock.readLock().unlock();
         releaseRowLock(lid);
       }
     } finally {