svn commit: r1064054 - in /hbase/branches/0.90: CHANGES.txt src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

[rawson]

Author: rawson
Date: Thu Jan 27 09:27:01 2011
New Revision: 1064054

URL: http://svn.apache.org/viewvc?rev=1064054&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/branches/0.90/CHANGES.txt
    
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

Modified: hbase/branches/0.90/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hbase/branches/0.90/CHANGES.txt?rev=1064054&r1=1064053&r2=1064054&view=diff
==============================================================================
--- hbase/branches/0.90/CHANGES.txt (original)
+++ hbase/branches/0.90/CHANGES.txt Thu Jan 27 09:27:01 2011
@@ -7,6 +7,8 @@ Release 0.90.1 - 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-3481  max seq id in flushed file can be larger than its correct 
+              value causing data loss during recovery
 
 
   IMPROVEMENTS

Modified: 
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: 
http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1064054&r1=1064053&r2=1064054&view=diff
==============================================================================
--- 
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
 (original)
+++ 
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
 Thu Jan 27 09:27:01 2011
@@ -1386,6 +1386,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);
@@ -1443,6 +1444,10 @@ public class HRegion implements HeapSize
             byteNow);
       }
 
+
+      this.updatesLock.readLock().lock();
+      locked = true;
+
       // ------------------------------------
       // STEP 3. Write to WAL
       // ----------------------------------
@@ -1474,6 +1479,9 @@ public class HRegion implements HeapSize
       success = true;
       return addedSize;
     } finally {
+      if (locked)
+        this.updatesLock.readLock().unlock();
+
       for (Integer toRelease : acquiredLocks) {
         releaseRowLock(toRelease);
       }
@@ -2986,6 +2994,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 :
@@ -3041,6 +3050,7 @@ public class HRegion implements HeapSize
         size = this.memstoreSize.addAndGet(size);
         flush = isFlushSize(size);
       } finally {
+        this.updatesLock.readLock().unlock();
         releaseRowLock(lid);
       }
     } finally {
@@ -3075,6 +3085,7 @@ public class HRegion implements HeapSize
     startRegionOperation();
     try {
       Integer lid = obtainRowLock(row);
+      this.updatesLock.readLock().lock();
       try {
         Store store = stores.get(family);
 
@@ -3113,6 +3124,7 @@ public class HRegion implements HeapSize
         size = this.memstoreSize.addAndGet(size);
         flush = isFlushSize(size);
       } finally {
+        this.updatesLock.readLock().unlock();
         releaseRowLock(lid);
       }
     } finally {