You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by en...@apache.org on 2015/10/30 02:47:03 UTC
[1/2] hbase git commit: HBASE-14689 Addendum and unit test for
HBASE-13471
Repository: hbase
Updated Branches:
refs/heads/0.98 5e22512cf -> 8c8b43433
refs/heads/branch-1.0 fe7cd0dee -> 4c4a43538
HBASE-14689 Addendum and unit test for HBASE-13471
Conflicts:
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/4c4a4353
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/4c4a4353
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/4c4a4353
Branch: refs/heads/branch-1.0
Commit: 4c4a4353898aef35016c342f997eba6e439cc039
Parents: fe7cd0d
Author: Enis Soztutar <en...@apache.org>
Authored: Wed Oct 28 18:16:32 2015 -0700
Committer: Enis Soztutar <en...@apache.org>
Committed: Thu Oct 29 18:38:27 2015 -0700
----------------------------------------------------------------------
.../hadoop/hbase/regionserver/HRegion.java | 8 +-
.../hadoop/hbase/regionserver/TestHRegion.java | 78 +++++++++++++++++---
2 files changed, 71 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/4c4a4353/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 45d6042..8a93079 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
@@ -2656,18 +2656,18 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
// 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(), shouldBlock);
+ rowLock = getRowLockInternal(mutation.getRow(), true);
} 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
- assert !shouldBlock : "Should never fail to get lock when blocking";
- break; // stop acquiring more rows for this batch
+ throw new IOException("Failed getting lock in batch put, row=" +
+ Bytes.toStringBinary(mutation.getRow()));
} else {
acquiredRowLocks.add(rowLock);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/4c4a4353/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 6669ccd..2a175ee 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
@@ -55,6 +55,11 @@ 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;
@@ -99,6 +104,7 @@ 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;
@@ -1392,9 +1398,12 @@ 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));
}
};
@@ -1402,9 +1411,6 @@ 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
@@ -1414,31 +1420,34 @@ 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();
- LOG.info("...waiting for put thread to sync 4th time");
- waitForCounter(source, "syncTimeNumOps", syncs + 4);
+ waitForCounter(source, "syncTimeNumOps", syncs + 1);
LOG.info("...joining on put thread");
ctx.stop();
@@ -6145,6 +6154,53 @@ 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);
+ }
+
private static HRegion initHRegion(byte[] tableName, String callingMethod,
byte[]... families) throws IOException {
return initHRegion(tableName, callingMethod, HBaseConfiguration.create(),
[2/2] hbase git commit: HBASE-14689 Addendum and unit test for
HBASE-13471
Posted by en...@apache.org.
HBASE-14689 Addendum and unit test for HBASE-13471
Conflicts:
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/8c8b4343
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/8c8b4343
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/8c8b4343
Branch: refs/heads/0.98
Commit: 8c8b4343323092291af1a9ce15088fa6883c9441
Parents: 5e22512
Author: Enis Soztutar <en...@apache.org>
Authored: Wed Oct 28 18:16:32 2015 -0700
Committer: Enis Soztutar <en...@apache.org>
Committed: Thu Oct 29 18:39:24 2015 -0700
----------------------------------------------------------------------
.../hadoop/hbase/regionserver/HRegion.java | 8 +-
.../hadoop/hbase/regionserver/TestHRegion.java | 78 +++++++++++++++++---
2 files changed, 71 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/8c8b4343/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 8036e2e..bf4c3b3 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
@@ -2484,18 +2484,18 @@ public class HRegion implements HeapSize { // , Writable{
// 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(), shouldBlock);
+ rowLock = getRowLockInternal(mutation.getRow(), true);
} 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
- assert !shouldBlock : "Should never fail to get lock when blocking";
- break; // stop acquiring more rows for this batch
+ throw new IOException("Failed getting lock in batch put, row=" +
+ Bytes.toStringBinary(mutation.getRow()));
} else {
acquiredRowLocks.add(rowLock);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/8c8b4343/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 cb210eb..16bb1b0 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
@@ -54,6 +54,11 @@ 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;
@@ -98,6 +103,7 @@ import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
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;
@@ -1120,9 +1126,12 @@ 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));
}
};
@@ -1130,9 +1139,6 @@ 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
@@ -1142,31 +1148,34 @@ 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();
- LOG.info("...waiting for put thread to sync 4th time");
- waitForCounter(source, "syncTimeNumOps", syncs + 4);
+ waitForCounter(source, "syncTimeNumOps", syncs + 1);
LOG.info("...joining on put thread");
ctx.stop();
@@ -5596,6 +5605,53 @@ 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);
+ }
+
private static HRegion initHRegion(byte[] tableName, String callingMethod,
byte[]... families) throws IOException {
return initHRegion(tableName, callingMethod, HBaseConfiguration.create(),