Fix #163 by only synchronizing on the probe suspended state, not the

entire RPC.

Signed-off-by: Chris Larsen <[email protected]>

src/HBaseClient.java

@@ -2802,26 +2802,24 @@ else if ((exists_rpc = nsred_rpcs.get(0)) != rpc) {
 
       // Stop here if this is a known NSRE and `rpc' is not our probe RPC that
       // is not suspended
-      synchronized (exists_rpc) {
-        if (known_nsre && exists_rpc != rpc && !exists_rpc.isSuspendedProbe()) {
-          if (size != nsre_high_watermark && size % NSRE_LOG_EVERY == 0) {
-            final String msg = "There are now " + size
-              + " RPCs pending due to NSRE on " + Bytes.pretty(region_name);
-            if (size + NSRE_LOG_EVERY < nsre_high_watermark) {
-              LOG.info(msg);  // First message logged at INFO level.
-            } else {
-              LOG.warn(msg);  // Last message logged with increased severity.
-            }
-          }
-          if (reject) {
-            rpc.callback(new PleaseThrottleException(size + " RPCs waiting on "
-              + Bytes.pretty(region_name) + " to come back online", e, rpc,
-              exists_rpc.getDeferred()));
+      if (known_nsre && exists_rpc != rpc && !exists_rpc.isSuspendedProbe()) {
+        if (size != nsre_high_watermark && size % NSRE_LOG_EVERY == 0) {
+          final String msg = "There are now " + size
+            + " RPCs pending due to NSRE on " + Bytes.pretty(region_name);
+          if (size + NSRE_LOG_EVERY < nsre_high_watermark) {
+            LOG.info(msg);  // First message logged at INFO level.
+          } else {
+            LOG.warn(msg);  // Last message logged with increased severity.
           }
-          return;  // This NSRE is already known and being handled.
         }
-        exists_rpc.setSuspendedProbe(false);
+        if (reject) {
+          rpc.callback(new PleaseThrottleException(size + " RPCs waiting on "
+            + Bytes.pretty(region_name) + " to come back online", e, rpc,
+            exists_rpc.getDeferred()));
+        }
+        return;  // This NSRE is already known and being handled.
       }
+      exists_rpc.setSuspendedProbe(false);
     }
 
     num_nsres.increment();

src/HBaseRpc.java

@@ -496,11 +496,11 @@ public void setProbe(boolean probe) {
    */
   private boolean suspended_probe = false;
 
-  boolean isSuspendedProbe() {
+  synchronized boolean isSuspendedProbe() {
     return suspended_probe;
   }
 
-  void setSuspendedProbe(boolean suspended_probe) {
+  synchronized void setSuspendedProbe(boolean suspended_probe) {
     this.suspended_probe = suspended_probe;
   }