hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e...@apache.org
Subject hbase git commit: Revert "HBASE-14689 Addendum and unit test for HBASE-13471"
Date Wed, 18 Nov 2015 04:42:47 GMT
Repository: hbase
Updated Branches:
  refs/heads/branch-1.1 c558f6df6 -> 214ea33d1


Revert "HBASE-14689 Addendum and unit test for HBASE-13471"

This reverts commit b15a05d7b611726ff9b2dd1a3c5c9448b41e1474.


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

Branch: refs/heads/branch-1.1
Commit: 214ea33d13a27de67b174e1fbeabb15e32b6943c
Parents: c558f6d
Author: Enis Soztutar <enis@apache.org>
Authored: Tue Nov 17 20:05:15 2015 -0800
Committer: Enis Soztutar <enis@apache.org>
Committed: Tue Nov 17 20:41:41 2015 -0800

----------------------------------------------------------------------
 .../hadoop/hbase/regionserver/HRegion.java      |  8 +-
 .../hadoop/hbase/regionserver/TestHRegion.java  | 78 +++-----------------
 2 files changed, 15 insertions(+), 71 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/214ea33d/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index 4558760..11c8945 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -2975,18 +2975,18 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
 
         // If we haven't got any rows in our batch, we should block to
         // get the next one.
+        boolean shouldBlock = numReadyToWrite == 0;
         RowLock rowLock = null;
         try {
-          rowLock = getRowLockInternal(mutation.getRow(), true);
+          rowLock = getRowLockInternal(mutation.getRow(), shouldBlock);
         } catch (IOException ioe) {
           LOG.warn("Failed getting lock in batch put, row="
             + Bytes.toStringBinary(mutation.getRow()), ioe);
-          throw ioe;
         }
         if (rowLock == null) {
           // We failed to grab another lock
-          throw new IOException("Failed getting lock in batch put, row=" +
-              Bytes.toStringBinary(mutation.getRow()));
+          assert !shouldBlock : "Should never fail to get lock when blocking";
+          break; // stop acquiring more rows for this batch
         } else {
           acquiredRowLocks.add(rowLock);
         }

http://git-wip-us.apache.org/repos/asf/hbase/blob/214ea33d/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
index e50fd9c..5add20e 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
@@ -56,11 +56,6 @@ import java.util.Map;
 import java.util.NavigableMap;
 import java.util.TreeMap;
 import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -105,7 +100,6 @@ import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Durability;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Increment;
-import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.RowMutations;
@@ -1508,12 +1502,9 @@ public class TestHRegion {
 
       MultithreadedTestUtil.TestContext ctx = new MultithreadedTestUtil.TestContext(CONF);
       final AtomicReference<OperationStatus[]> retFromThread = new AtomicReference<OperationStatus[]>();
-      final CountDownLatch startingPuts = new CountDownLatch(1);
-      final CountDownLatch startingClose = new CountDownLatch(1);
       TestThread putter = new TestThread(ctx) {
         @Override
         public void doWork() throws IOException {
-          startingPuts.countDown();
           retFromThread.set(region.batchMutate(puts));
         }
       };
@@ -1521,6 +1512,9 @@ public class TestHRegion {
       ctx.addThread(putter);
       ctx.startThreads();
 
+      LOG.info("...waiting for put thread to sync 1st time");
+      waitForCounter(source, "syncTimeNumOps", syncs + 1);
+
       // Now attempt to close the region from another thread.  Prior to HBASE-12565
       // this would cause the in-progress batchMutate operation to to fail with
       // exception because it use to release and re-acquire the close-guard lock
@@ -1530,34 +1524,31 @@ public class TestHRegion {
         @Override
         public void run() {
           try {
-            startingPuts.await();
-            // Give some time for the batch mutate to get in.
-            // We don't want to race with the mutate
-            Thread.sleep(10);
-            startingClose.countDown();
             HRegion.closeHRegion(region);
           } catch (IOException e) {
             throw new RuntimeException(e);
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
           }
         }
       };
       regionCloseThread.start();
 
-      startingClose.await();
-      startingPuts.await();
-      Thread.sleep(100);
       LOG.info("...releasing row lock 1, which should let put thread continue");
       rowLock1.release();
 
+      LOG.info("...waiting for put thread to sync 2nd time");
+      waitForCounter(source, "syncTimeNumOps", syncs + 2);
+
       LOG.info("...releasing row lock 2, which should let put thread continue");
       rowLock2.release();
 
+      LOG.info("...waiting for put thread to sync 3rd time");
+      waitForCounter(source, "syncTimeNumOps", syncs + 3);
+
       LOG.info("...releasing row lock 3, which should let put thread continue");
       rowLock3.release();
 
-      waitForCounter(source, "syncTimeNumOps", syncs + 1);
+      LOG.info("...waiting for put thread to sync 4th time");
+      waitForCounter(source, "syncTimeNumOps", syncs + 4);
 
       LOG.info("...joining on put thread");
       ctx.stop();
@@ -6381,53 +6372,6 @@ public class TestHRegion {
       qual2, 0, qual2.length));
   }
 
-  @Test(timeout = 30000)
-  public void testBatchMutateWithWrongRegionException() throws IOException, InterruptedException
{
-    final byte[] a = Bytes.toBytes("a");
-    final byte[] b = Bytes.toBytes("b");
-    final byte[] c = Bytes.toBytes("c"); // exclusive
-
-    int prevLockTimeout = CONF.getInt("hbase.rowlock.wait.duration", 30000);
-    CONF.setInt("hbase.rowlock.wait.duration", 3000);
-    final HRegion region = initHRegion(tableName, a, c, name.getMethodName(), CONF, false,
fam1);
-
-    Mutation[] mutations = new Mutation[] {
-        new Put(a).addImmutable(fam1, null, null),
-        new Put(c).addImmutable(fam1, null, null), // this is outside the region boundary
-        new Put(b).addImmutable(fam1, null, null),
-    };
-
-    OperationStatus[] status = region.batchMutate(mutations);
-    assertEquals(status[0].getOperationStatusCode(), OperationStatusCode.SUCCESS);
-    assertEquals(status[1].getOperationStatusCode(), OperationStatusCode.SANITY_CHECK_FAILURE);
-    assertEquals(status[2].getOperationStatusCode(), OperationStatusCode.SUCCESS);
-
-    // test with a leaked row lock
-    ExecutorService exec = Executors.newSingleThreadExecutor();
-    exec.submit(new Callable<Void>() {
-      @Override
-      public Void call() throws Exception {
-        region.getRowLock(b);
-        return null;
-      }
-    });
-    exec.shutdown();
-    exec.awaitTermination(30, TimeUnit.SECONDS);
-
-    mutations = new Mutation[] {
-        new Put(a).addImmutable(fam1, null, null),
-        new Put(b).addImmutable(fam1, null, null),
-    };
-
-    try {
-      status = region.batchMutate(mutations);
-      fail("Failed to throw exception");
-    } catch (IOException expected) {
-    }
-
-    CONF.setInt("hbase.rowlock.wait.duration", prevLockTimeout);
-  }
-
   static HRegion initHRegion(byte[] tableName, String callingMethod,
       byte[]... families) throws IOException {
     return initHRegion(tableName, callingMethod, HBaseConfiguration.create(),


Mime
View raw message