Harden Phoenix - distributed big link list and atomicity tests over phoenix

phoenix-core/src/it/java/org/apache/phoenix/end2end/MultithreadedTestUtil.java

@@ -0,0 +1,182 @@
+package org.apache.phoenix.end2end;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Based on HBase's testing framework.
+ * Source: hbase-rel-1.1.5/hbase-server/src/test/java/org/apache/hadoop/hbase/MultithreadedTestUtil.java
+ */
+public class MultithreadedTestUtil {
+ private static final Log LOG = LogFactory.getLog(MultithreadedTestUtil.class);
+
+ public static class TestContext {
+ private final Configuration conf;
+ private Throwable err = null;
+ private boolean stopped = false;
+ private int threadDoneCount = 0;
+ private Set<TestThread> testThreads = new HashSet<TestThread>();
+
+ public TestContext(Configuration configuration) {
+ this.conf = configuration;
+ }
+
+ protected Configuration getConf() {
+ return conf;
+ }
+
+ public synchronized boolean shouldRun() {
+ return !stopped && err == null;
+ }
+
+ public void addThread(TestThread t) {
+ testThreads.add(t);
+ }
+
+ public void startThreads() {
+ for (TestThread t : testThreads) {
+ t.start();
+ }
+ }
+
+ public void waitFor(long millis) throws Exception {
+ long endTime = System.currentTimeMillis() + millis;
+ while (!stopped) {
+ long left = endTime - System.currentTimeMillis();
+ if (left <= 0) break;
+ synchronized (this) {
+ checkException();
+ wait(left);
+ }
+ }
+ }
+ private synchronized void checkException() throws Exception {
+ if (err != null) {
+ throw new RuntimeException("Deferred", err);
+ }
+ }
+
+ public synchronized void threadFailed(Throwable t) {
+ if (err == null) err = t;
+ LOG.error("Failed!", err);
+ notify();
+ }
+
+ public synchronized void threadDone() {
+ threadDoneCount++;
+ }
+
+ public boolean removeAllThreads(){
+ if(shouldRun()){
+ return false;
+ } else {
+ testThreads.clear();
+ return true;
+ }
+ }
+
+ public void setStopFlag(boolean s) throws Exception {
+ synchronized (this) {
+ stopped = s;
+ }
+ }
+
+ public void stop() throws Exception {
+ synchronized (this) {
+ stopped = true;
+ }
+ for (TestThread t : testThreads) {
+ t.join();
+ }
+ checkException();
+ }
+ }
+
+ /**
+ * A thread that can be added to a test context, and properly
+ * passes exceptions through.
+ */
+ public static abstract class TestThread extends Thread {
+ protected final TestContext ctx;
+ protected boolean stopped;
+
+ public TestThread(TestContext ctx) {
+ this.ctx = ctx;
+ }
+
+ public void run() {
+ try {
+ doWork();
+ } catch (Throwable t) {
+ ctx.threadFailed(t);
+ }
+ ctx.threadDone();
+ }
+
+ public abstract void doWork() throws Exception;
+
+ protected void stopTestThread() {
+ this.stopped = true;
+ }
+ }
+
+ /**
+ * A test thread that performs a repeating operation.
+ */
+ public static abstract class RepeatingTestThread extends TestThread {
+ public RepeatingTestThread(TestContext ctx) {
+ super(ctx);
+ }
+
+ public final void doWork() throws Exception {
+ while (ctx.shouldRun() && !stopped) {
+ doAnAction();
+ }
+ }
+
+ public abstract void doAnAction() throws Exception;
+ }
+
+ /**
+ * Verify that no assertions have failed inside a future.
+ * Used for unit tests that spawn threads. E.g.,
+ * <p>
+ * <code>
+ * List<Future<Void>> results = Lists.newArrayList();
+ * Future<Void> f = executor.submit(new Callable<Void> {
+ * public Void call() {
+ * assertTrue(someMethod());
+ * }
+ * });
+ * results.add(f);
+ * assertOnFutures(results);
+ * </code>
+ * @param threadResults A list of futures
+ * @param <T>
+ * @throws InterruptedException If interrupted when waiting for a result
+ * from one of the futures
+ * @throws ExecutionException If an exception other than AssertionError
+ * occurs inside any of the futures
+ */
+ public static <T> void assertOnFutures(List<Future<T>> threadResults)
+ throws InterruptedException, ExecutionException {
+ for (Future<T> threadResult : threadResults) {
+ try {
+ threadResult.get();
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof AssertionError) {
+ throw (AssertionError) e.getCause();
+ }
+ throw e;
+ }
+ }
+ }
+}

phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixHBaseSuiteAtomicityIT.java

@@ -0,0 +1,141 @@
+package org.apache.phoenix.end2end;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.*;
+
+/**
+ * Test case that uses multiple threads to read and write rows
+ * from/into a table, verifying that reads never see partially-complete writes.
+ * Meant to be a port of HBase's atomicity test.
+ * Source: https://github.com/apache/hbase/blob/master/hbase-server/src/test/java/org/apache/hadoop/hbase/TestAcidGuarantees.java
+ */
+public class PhoenixHBaseSuiteAtomicityIT extends BaseHBaseManagedTimeIT {
+ protected static final Log LOG = LogFactory.getLog(PhoenixHBaseSuiteAtomicityIT.class);
+
+ //Two arbitrary values for writers to randomly alternate between.
+ final int DataVal1 = 1023; //2^10 - 1
+ final int DataVal2 = 33; //2^5 - 1
+
+ final String TableName = "TestAcidGuarantees";
+ final String TestTable = "CREATE TABLE IF NOT EXISTS " + TableName +
+ "(a_id INTEGER NOT NULL, " +
+ "a_data INTEGER, " +
+ "CONSTRAINT my_pk PRIMARY KEY (a_id))";
+
+ /**
+ * Thread that does random full-row writes into a table.
+ */
+ private class RandomWriter extends MultithreadedTestUtil.RepeatingTestThread {
+ Random rand = new Random();
+ Connection conn;
+ String tableName;
+ int data;
+ AtomicLong numWritten = new AtomicLong();
+
+ private RandomWriter(MultithreadedTestUtil.TestContext ctx, Connection conn, String tableName) throws IOException {
+ super(ctx);
+ this.conn = conn;
+ this.tableName = tableName;
+ }
+
+ public void doAnAction() throws Exception {
+ if(rand.nextBoolean()){
+ data = DataVal1; //2^10 - 1
+ } else {
+ data = DataVal2; //2^5 + 1
+ }
+
+ // Pick a random row to write into
+ String randomID = Integer.toString(rand.nextInt(50));
+ synchronized (conn){
+ conn.createStatement().execute("UPSERT INTO " + tableName +
+ " VALUES ("+randomID+","+Integer.toString(data)+")");
+ conn.commit();
+ }
+ numWritten.getAndIncrement();
+ }
+
+ }
+
+ /**
+ * Thread that does scans of a table, looking for partially
+ * completed rows.
+ */
+ private class RandomReader extends MultithreadedTestUtil.RepeatingTestThread {
+ Connection conn;
+ String tableName;
+ int idOffset;
+ int idRange;
+ AtomicLong numRead = new AtomicLong();
+
+ private RandomReader(MultithreadedTestUtil.TestContext ctx, Connection conn, String tableName, int idOffset, int idRange) throws IOException {
+ super(ctx);
+ this.conn = conn;
+ this.tableName = tableName;
+ this.idOffset = idOffset;
+ this.idRange = idRange;
+ }
+
+ public void doAnAction() throws Exception {
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT a_data FROM " + tableName +
+ " WHERE a_id >= " + Integer.toString(idOffset) +
+ " AND a_id < " + Integer.toString(idOffset + idRange));
+
+ while (rs.next()) {
+ int thisValue = rs.getInt(1);
+ assertTrue(thisValue == DataVal1 || thisValue == DataVal2);
+ numRead.getAndIncrement();
+ }
+ }
+
+ }
+
+ @Test
+ public void testScanAtomicity() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+ String sql = TestTable;
+
+ PreparedStatement statement = conn.prepareStatement(sql);
+ statement.execute();
+ synchronized (conn){
+ conn.commit();
+ }
+
+ MultithreadedTestUtil.TestContext ctx = new MultithreadedTestUtil.TestContext(getTestClusterConfig());
+ List<RandomWriter> testWriters = new ArrayList<RandomWriter>();
+ List<RandomReader> testReaders = new ArrayList<RandomReader>();
+ for(int i = 0; i<5; ++i){
+ testWriters.add(new RandomWriter(ctx, conn, TableName));
+ ctx.addThread(testWriters.get(i));
+ }
+ for(int i=0; i<4; ++i){
+ testReaders.add(new RandomReader(ctx, conn, TableName, (25*i)%50, 25));
+ ctx.addThread(testReaders.get(i));
+ }
+ ctx.startThreads();
+ ctx.waitFor(3000);
+ ctx.stop();
+
+ LOG.info("Finished test.");
+ LOG.info("Finished test. Writers:");
+ for (RandomWriter writer : testWriters) {
+ LOG.info(" wrote " + writer.numWritten.get());
+ }
+ LOG.info("Readers:");
+ for (RandomReader reader : testReaders) {
+ LOG.info(" read " + reader.numRead.get());
+ }
+
+ }
+}