From bbe297121d127ef14863d8763d20fcedf198904d Mon Sep 17 00:00:00 2001
From: Zhihong Yu <tedyu@apache.org>
Date: Thu, 23 Aug 2012 13:48:39 +0000
Subject: [PATCH] HBASE-5329 addRowLock() may allocate duplicate lock id,
 causing the client to be blocked (Ian Varley)

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1376489 13f79535-47bb-0310-9956-ffa450edef68
---
 .../hadoop/hbase/regionserver/HRegionServer.java  | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
index ac6f1550351..3ba1e24608c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -51,6 +51,7 @@ import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -320,7 +321,7 @@ public class  HRegionServer implements ClientProtocol,
   // Port we put up the webui on.
   protected int webuiport = -1;
 
-  Map<String, Integer> rowlocks = new ConcurrentHashMap<String, Integer>();
+  ConcurrentMap<String, Integer> rowlocks = new ConcurrentHashMap<String, Integer>();
 
   // A state before we go into stopped state.  At this stage we're closing user
   // space regions.
@@ -2753,11 +2754,13 @@ public class  HRegionServer implements ClientProtocol,
     return this.fsOk;
   }
 
-  protected long addRowLock(Integer r, HRegion region)
-      throws LeaseStillHeldException {
-    long lockId = nextLong();
-    String lockName = String.valueOf(lockId);
-    rowlocks.put(lockName, r);
+  protected long addRowLock(Integer r, HRegion region) throws LeaseStillHeldException {
+    String lockName = null;
+    long lockId;
+    do {
+      lockId = nextLong();
+      lockName = String.valueOf(lockId);
+    } while (rowlocks.putIfAbsent(lockName, r) != null);
     this.leases.createLease(lockName, this.rowLockLeaseTimeoutPeriod, new RowLockListener(lockName,
         region));
     return lockId;