svn commit: r1307631 [2/2] - in /hbase/branches/0.89-fb: ./ src/main/java/org/apache/hadoop/hbase/ src/main/java/org/apache/hadoop/hbase/regionserver/ src/main/java/org/apache/hadoop/hbase/util/ src/main/java/org/apache/hadoop/hbase/zookeeper/ src/test...

[mb...@apache.org]

Added: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadParallel.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadParallel.java?rev=1307631&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadParallel.java (added)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadParallel.java Fri Mar 30 21:49:13 2012
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hadoop.hbase.util;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+/**
+ * A write/read/verify load test on a mini HBase cluster. Tests reading
+ * and writing at the same time.
+ */
+@RunWith(Parameterized.class)
+public class TestMiniClusterLoadParallel
+    extends TestMiniClusterLoadSequential {
+
+  public TestMiniClusterLoadParallel(boolean isMultiPut,
+      DataBlockEncoding encoding) {
+    super(isMultiPut, encoding);
+  }
+
+  @Test(timeout=TIMEOUT_MS)
+  public void loadTest() throws Exception {
+    prepareForLoadTest();
+
+    readerThreads.linkToWriter(writerThreads);
+
+    writerThreads.start(0, numKeys, NUM_THREADS);
+    readerThreads.start(0, numKeys, NUM_THREADS);
+
+    writerThreads.waitForFinish();
+    readerThreads.waitForFinish();
+
+    assertEquals(0, writerThreads.getNumWriteFailures());
+    assertEquals(0, readerThreads.getNumReadFailures());
+    assertEquals(0, readerThreads.getNumReadErrors());
+    assertEquals(numKeys, readerThreads.getNumUniqueKeysVerified());
+  }
+
+}

Added: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadSequential.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadSequential.java?rev=1307631&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadSequential.java (added)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/util/TestMiniClusterLoadSequential.java Fri Mar 30 21:49:13 2012
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hadoop.hbase.util;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
+import org.apache.hadoop.hbase.io.hfile.Compression;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+/**
+ * A write/read/verify load test on a mini HBase cluster. Tests reading
+ * and then writing.
+ */
+@RunWith(Parameterized.class)
+public class TestMiniClusterLoadSequential {
+
+  private static final Log LOG = LogFactory.getLog(
+      TestMiniClusterLoadSequential.class);
+
+  protected static final byte[] TABLE = Bytes.toBytes("load_test_tbl");
+  protected static final byte[] CF = Bytes.toBytes("load_test_cf");
+  protected static final int NUM_THREADS = 8;
+  protected static final int NUM_RS = 2;
+  protected static final int TIMEOUT_MS = 120000;
+  protected static final HBaseTestingUtility TEST_UTIL =
+      new HBaseTestingUtility();
+
+  protected final Configuration conf = TEST_UTIL.getConfiguration();
+  protected final boolean isMultiPut;
+  protected final DataBlockEncoding dataBlockEncoding;
+
+  protected MultiThreadedWriter writerThreads;
+  protected MultiThreadedReader readerThreads;
+  protected int numKeys;
+
+  protected Compression.Algorithm compression = Compression.Algorithm.NONE;
+
+  public TestMiniClusterLoadSequential(boolean isMultiPut,
+      DataBlockEncoding dataBlockEncoding) {
+    this.isMultiPut = isMultiPut;
+    this.dataBlockEncoding = dataBlockEncoding;
+    conf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 1024 * 1024);
+  }
+
+  @Parameters
+  public static Collection<Object[]> parameters() {
+    List<Object[]> parameters = new ArrayList<Object[]>();
+    for (boolean multiPut : new boolean[]{false, true}) {
+      for (DataBlockEncoding dataBlockEncoding : new DataBlockEncoding[] {
+          DataBlockEncoding.NONE, DataBlockEncoding.PREFIX }) {
+        parameters.add(new Object[]{multiPut, dataBlockEncoding});
+      }
+    }
+    return parameters;
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    LOG.debug("Test setup: isMultiPut=" + isMultiPut);
+    TEST_UTIL.startMiniCluster(1, NUM_RS);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    LOG.debug("Test teardown: isMultiPut=" + isMultiPut);
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Test(timeout=TIMEOUT_MS)
+  public void loadTest() throws Exception {
+    prepareForLoadTest();
+    runLoadTestOnExistingTable();
+  }
+
+  protected void runLoadTestOnExistingTable() throws IOException {
+    writerThreads.start(0, numKeys, NUM_THREADS);
+    writerThreads.waitForFinish();
+    assertEquals(0, writerThreads.getNumWriteFailures());
+
+    readerThreads.start(0, numKeys, NUM_THREADS);
+    readerThreads.waitForFinish();
+    assertEquals(0, readerThreads.getNumReadFailures());
+    assertEquals(0, readerThreads.getNumReadErrors());
+    assertEquals(numKeys, readerThreads.getNumKeysVerified());
+  }
+
+  protected void prepareForLoadTest() throws IOException {
+    LOG.info("Starting load test: dataBlockEncoding=" + dataBlockEncoding +
+        ", isMultiPut=" + isMultiPut);
+    numKeys = numKeys();
+    HBaseAdmin admin = new HBaseAdmin(conf);
+    while (admin.getClusterStatus().getServers() < NUM_RS) {
+      LOG.info("Sleeping until " + NUM_RS + " RSs are online");
+      Threads.sleepWithoutInterrupt(1000);
+    }
+    admin.close();
+
+    int numRegions = HBaseTestingUtility.createPreSplitLoadTestTable(conf,
+        TABLE, CF, compression, dataBlockEncoding);
+
+    TEST_UTIL.waitUntilAllRegionsAssigned(numRegions);
+
+    writerThreads = new MultiThreadedWriter(conf, TABLE, CF);
+    writerThreads.setMultiPut(isMultiPut);
+    readerThreads = new MultiThreadedReader(conf, TABLE, CF, 100);
+  }
+
+  protected int numKeys() {
+    return 10000;
+  }
+
+  protected HColumnDescriptor getColumnDesc(HBaseAdmin admin)
+      throws TableNotFoundException, IOException {
+    return admin.getTableDescriptor(TABLE).getFamily(CF);
+  }
+
+}