hbase git commit: HBASE-17904 Get runs into NoSuchElementException when using Read Replica, with hbase. ipc.client.specificThreadForWriting to be true and hbase.rpc.client.impl to be org.apache.hadoop.hbase.ipc.RpcClientImpl

[zh...@apache.org]

Repository: hbase
Updated Branches:
  refs/heads/branch-1 a2e4e0a1e -> cdda1d030


HBASE-17904 Get runs into NoSuchElementException when using Read Replica, with hbase. ipc.client.specificThreadForWriting to be true and hbase.rpc.client.impl to be org.apache.hadoop.hbase.ipc.RpcClientImpl

Signed-off-by: zhangduo <zh...@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/cdda1d03
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/cdda1d03
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/cdda1d03

Branch: refs/heads/branch-1
Commit: cdda1d0302238d7309048a01a667da1dedf67ab3
Parents: a2e4e0a
Author: huaxiang sun <hs...@cloudera.com>
Authored: Fri Apr 28 06:53:37 2017 +0800
Committer: zhangduo <zh...@apache.org>
Committed: Fri Apr 28 06:53:37 2017 +0800

----------------------------------------------------------------------
 .../hadoop/hbase/ipc/BlockingRpcConnection.java |  4 +-
 .../hbase/client/TestReplicaWithCluster.java    | 50 ++++++++++++++++++++
 2 files changed, 52 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/cdda1d03/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java
index 4dc121c..32c4ab9 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java
@@ -152,7 +152,7 @@ class BlockingRpcConnection extends RpcConnection implements Runnable {
     }
 
     public void remove(Call call) {
-      callsToWrite.remove();
+      callsToWrite.remove(call);
       // By removing the call from the expected call list, we make the list smaller, but
       // it means as well that we don't know how many calls we cancelled.
       calls.remove(call.id);
@@ -727,4 +727,4 @@ class BlockingRpcConnection extends RpcConnection implements Runnable {
   public synchronized boolean isActive() {
     return thread != null;
   }
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/cdda1d03/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java
index 617a8f7..892cff2 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java
@@ -521,4 +521,54 @@ public class TestReplicaWithCluster {
       HTU.deleteTable(hdt.getTableName());
     }
   }
+
+  @Test
+  public void testReplicaGetWithRpcClientImpl() throws IOException {
+    HTU.getConfiguration().setBoolean("hbase.ipc.client.specificThreadForWriting", true);
+    HTU.getConfiguration().set("hbase.rpc.client.impl", "org.apache.hadoop.hbase.ipc.RpcClientImpl");
+    // Create table then get the single region for our new table.
+    HTableDescriptor hdt = HTU.createTableDescriptor("testReplicaGetWithRpcClientImpl");
+    hdt.setRegionReplication(NB_SERVERS);
+    hdt.addCoprocessor(SlowMeCopro.class.getName());
+
+    try {
+      Table table = HTU.createTable(hdt, new byte[][] { f }, null);
+
+      Put p = new Put(row);
+      p.addColumn(f, row, row);
+      table.put(p);
+
+      // Flush so it can be picked by the replica refresher thread
+      HTU.flush(table.getName());
+
+      // Sleep for some time until data is picked up by replicas
+      try {
+        Thread.sleep(2 * REFRESH_PERIOD);
+      } catch (InterruptedException e1) {
+        LOG.error(e1);
+      }
+
+      try {
+        // Create the new connection so new config can kick in
+        Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
+        Table t = connection.getTable(hdt.getTableName());
+
+        // But if we ask for stale we will get it
+        SlowMeCopro.cdl.set(new CountDownLatch(1));
+        Get g = new Get(row);
+        g.setConsistency(Consistency.TIMELINE);
+        Result r = t.get(g);
+        Assert.assertTrue(r.isStale());
+        SlowMeCopro.cdl.get().countDown();
+      } finally {
+        SlowMeCopro.cdl.get().countDown();
+        SlowMeCopro.sleepTime.set(0);
+      }
+    } finally {
+      HTU.getConfiguration().unset("hbase.ipc.client.specificThreadForWriting");
+      HTU.getConfiguration().unset("hbase.rpc.client.impl");
+      HTU.getHBaseAdmin().disableTable(hdt.getTableName());
+      HTU.deleteTable(hdt.getTableName());
+    }
+  }
 }