You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2013/04/24 21:56:37 UTC
svn commit: r1471622 - in
/hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver:
TestAtomicOperation.java TestHBase7051.java
Author: stack
Date: Wed Apr 24 19:56:37 2013
New Revision: 1471622
URL: http://svn.apache.org/r1471622
Log:
HBASE-7192 Move TestHBase7051.java into TestAtomicOperation.java
Removed:
hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHBase7051.java
Modified:
hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestAtomicOperation.java
Modified: hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestAtomicOperation.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestAtomicOperation.java?rev=1471622&r1=1471621&r2=1471622&view=diff
==============================================================================
--- hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestAtomicOperation.java (original)
+++ hbase/branches/0.95/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestAtomicOperation.java Wed Apr 24 19:56:37 2013
@@ -23,19 +23,47 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.*;
-import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestCase;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.MediumTests;
+import org.apache.hadoop.hbase.MultithreadedTestUtil;
+import org.apache.hadoop.hbase.MultithreadedTestUtil.TestContext;
+import org.apache.hadoop.hbase.MultithreadedTestUtil.TestThread;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+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;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.BinaryComparator;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.io.HeapSize;
+import org.apache.hadoop.hbase.regionserver.wal.HLog;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
+import org.apache.hadoop.hbase.util.Pair;
import org.junit.experimental.categories.Category;
+import com.google.common.collect.Lists;
+
/**
* Testing of HRegion.incrementColumnValue, HRegion.increment,
@@ -473,6 +501,149 @@ public class TestAtomicOperation extends
this.failures = failures;
}
}
+
+ private static CountDownLatch latch = new CountDownLatch(1);
+ private enum TestStep {
+ INIT, // initial put of 10 to set value of the cell
+ PUT_STARTED, // began doing a put of 50 to cell
+ PUT_COMPLETED, // put complete (released RowLock, but may not have advanced MVCC).
+ CHECKANDPUT_STARTED, // began checkAndPut: if 10 -> 11
+ CHECKANDPUT_COMPLETED // completed checkAndPut
+ // NOTE: at the end of these steps, the value of the cell should be 50, not 11!
+ }
+ private static volatile TestStep testStep = TestStep.INIT;
+ private final String family = "f1";
+
+ /**
+ * Test written as a verifier for HBASE-7051, CheckAndPut should properly read
+ * MVCC.
+ *
+ * Moved into TestAtomicOperation from its original location, TestHBase7051
+ */
+ public void testPutAndCheckAndPutInParallel() throws Exception {
+
+ final String tableName = "testPutAndCheckAndPut";
+ Configuration conf = HBaseConfiguration.create();
+ conf.setClass(HConstants.REGION_IMPL, MockHRegion.class, HeapSize.class);
+ final MockHRegion region = (MockHRegion) TestHRegion.initHRegion(
+ Bytes.toBytes(tableName), tableName, conf, Bytes.toBytes(family));
+
+ List<Pair<Mutation, Integer>> putsAndLocks = Lists.newArrayList();
+ Put[] puts = new Put[1];
+ Put put = new Put(Bytes.toBytes("r1"));
+ put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("10"));
+ puts[0] = put;
+ Pair<Mutation, Integer> pair = new Pair<Mutation, Integer>(puts[0], null);
+
+ putsAndLocks.add(pair);
+
+ region.batchMutate(putsAndLocks.toArray(new Pair[0]));
+ MultithreadedTestUtil.TestContext ctx =
+ new MultithreadedTestUtil.TestContext(conf);
+ ctx.addThread(new PutThread(ctx, region));
+ ctx.addThread(new CheckAndPutThread(ctx, region));
+ ctx.startThreads();
+ while (testStep != TestStep.CHECKANDPUT_COMPLETED) {
+ Thread.sleep(100);
+ }
+ ctx.stop();
+ Scan s = new Scan();
+ RegionScanner scanner = region.getScanner(s);
+ List<KeyValue> results = new ArrayList<KeyValue>();
+ scanner.next(results, 2);
+ for (KeyValue keyValue : results) {
+ assertEquals("50",Bytes.toString(keyValue.getValue()));
+ }
+
+ }
+
+ private class PutThread extends TestThread {
+ private MockHRegion region;
+ PutThread(TestContext ctx, MockHRegion region) {
+ super(ctx);
+ this.region = region;
+ }
+
+ public void doWork() throws Exception {
+ List<Pair<Mutation, Integer>> putsAndLocks = Lists.newArrayList();
+ Put[] puts = new Put[1];
+ Put put = new Put(Bytes.toBytes("r1"));
+ put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("50"));
+ puts[0] = put;
+ Pair<Mutation, Integer> pair = new Pair<Mutation, Integer>(puts[0], null);
+ putsAndLocks.add(pair);
+ testStep = TestStep.PUT_STARTED;
+ region.batchMutate(putsAndLocks.toArray(new Pair[0]));
+ }
+ }
+
+ private class CheckAndPutThread extends TestThread {
+ private MockHRegion region;
+ CheckAndPutThread(TestContext ctx, MockHRegion region) {
+ super(ctx);
+ this.region = region;
+ }
+
+ public void doWork() throws Exception {
+ Put[] puts = new Put[1];
+ Put put = new Put(Bytes.toBytes("r1"));
+ put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("11"));
+ puts[0] = put;
+ while (testStep != TestStep.PUT_COMPLETED) {
+ Thread.sleep(100);
+ }
+ testStep = TestStep.CHECKANDPUT_STARTED;
+ region.checkAndMutate(Bytes.toBytes("r1"), Bytes.toBytes(family), Bytes.toBytes("q1"),
+ CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("10")), put, true);
+ testStep = TestStep.CHECKANDPUT_COMPLETED;
+ }
+ }
-}
+ public static class MockHRegion extends HRegion {
+ public MockHRegion(Path tableDir, HLog log, FileSystem fs, Configuration conf,
+ final HRegionInfo regionInfo, final HTableDescriptor htd, RegionServerServices rsServices) {
+ super(tableDir, log, fs, conf, regionInfo, htd, rsServices);
+ }
+
+ @Override
+ public void releaseRowLock(Integer lockId) {
+ if (testStep == TestStep.INIT) {
+ super.releaseRowLock(lockId);
+ return;
+ }
+
+ if (testStep == TestStep.PUT_STARTED) {
+ try {
+ testStep = TestStep.PUT_COMPLETED;
+ super.releaseRowLock(lockId);
+ // put has been written to the memstore and the row lock has been released, but the
+ // MVCC has not been advanced. Prior to fixing HBASE-7051, the following order of
+ // operations would cause the non-atomicity to show up:
+ // 1) Put releases row lock (where we are now)
+ // 2) CheckAndPut grabs row lock and reads the value prior to the put (10)
+ // because the MVCC has not advanced
+ // 3) Put advances MVCC
+ // So, in order to recreate this order, we wait for the checkAndPut to grab the rowLock
+ // (see below), and then wait some more to give the checkAndPut time to read the old
+ // value.
+ latch.await();
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ else if (testStep == TestStep.CHECKANDPUT_STARTED) {
+ super.releaseRowLock(lockId);
+ }
+ }
+
+ @Override
+ public Integer getLock(Integer lockid, byte[] row, boolean waitForLock) throws IOException {
+ if (testStep == TestStep.CHECKANDPUT_STARTED) {
+ latch.countDown();
+ }
+ return super.getLock(lockid, row, waitForLock);
+ }
+ }
+}
\ No newline at end of file