diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MultithreadedTestUtil.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MultithreadedTestUtil.java new file mode 100644 index 00000000000..72d1acd545c --- /dev/null +++ b/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 testThreads = new HashSet(); + + 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., + *

+ * + * List> results = Lists.newArrayList(); + * Future f = executor.submit(new Callable { + * public Void call() { + * assertTrue(someMethod()); + * } + * }); + * results.add(f); + * assertOnFutures(results); + * + * @param threadResults A list of futures + * @param + * @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 void assertOnFutures(List> threadResults) + throws InterruptedException, ExecutionException { + for (Future threadResult : threadResults) { + try { + threadResult.get(); + } catch (ExecutionException e) { + if (e.getCause() instanceof AssertionError) { + throw (AssertionError) e.getCause(); + } + throw e; + } + } + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixHBaseSuiteAtomicityIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixHBaseSuiteAtomicityIT.java new file mode 100644 index 00000000000..8b71e137405 --- /dev/null +++ b/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 testWriters = new ArrayList(); + List testReaders = new ArrayList(); + 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()); + } + + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixHBaseSuiteLinkedListIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixHBaseSuiteLinkedListIT.java new file mode 100644 index 00000000000..e0af4ae9e66 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PhoenixHBaseSuiteLinkedListIT.java @@ -0,0 +1,182 @@ +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.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * This test case is based off of HBase's BigLinkedList IT. + * Source: https://github.com/apache/hbase/blob/master/hbase-it/src/test/java/org/apache/hadoop/hbase/test/IntegrationTestBigLinkedList.java + * + * This test creates 250 000 nodes and connects them into a 10 000 linked lists. Once the linking process is finished, + * the test runs through them and verifies that there are no broken links or holes, since these would + * indicate data loss. + * + * Further Work: The idea is to have this test running while a background process begins shutting down system + * modules, thus testing the robustness of the system. + * + * Note: If more processing power is available, the test can be scaled up. + */ +public class PhoenixHBaseSuiteLinkedListIT extends BaseHBaseManagedTimeIT { + protected static final Log LOG = LogFactory.getLog(PhoenixHBaseSuiteLinkedListIT.class); + final String tableName = "BigLinkedList"; + + final String TestTable = "CREATE TABLE IF NOT EXISTS " + tableName + + "(a_key INTEGER NOT NULL, " + + "a_prev INTEGER NOT NULL, " + + "CONSTRAINT my_pk PRIMARY KEY (a_key, a_prev))"; + + /** + * Thread that does full-row writes into a table. + */ + private class NodeWriter extends MultithreadedTestUtil.TestThread { + Connection conn; + String tableName; + int keyOffset; + int keyRange; + AtomicInteger prevKeyCounter; + + private NodeWriter(MultithreadedTestUtil.TestContext ctx, Connection conn, String tableName, int keyOffset, int keyRange, AtomicInteger prevKeyCounter) throws IOException { + super(ctx); + this.conn = conn; + this.tableName = tableName; + this.keyOffset = keyOffset; + this.keyRange = keyRange; + this.prevKeyCounter = prevKeyCounter; + } + + public void doWork() throws SQLException { + for (int n = keyOffset; n < (keyOffset + keyRange); ++n) { + int prev = prevKeyCounter.getAndIncrement(); + synchronized (conn) { + conn.createStatement().execute("UPSERT INTO " + tableName + + " VALUES (" + Integer.toString(n) + "," + Integer.toString(prev) + ")"); + } + } + synchronized (conn) { + conn.commit(); + } + } + } + + /** + * Thread that does single-row reads in a table, looking for partially + * + * completed rows. + */ + private class ListVerifier extends MultithreadedTestUtil.TestThread { + Connection conn; + String tableName; + int keyOffset; + int keyRange; + int listLength; + int numLists; + + private ListVerifier(MultithreadedTestUtil.TestContext ctx, Connection conn, String tableName, int keyOffset, + int keyRange, int listLength, int numLists) throws IOException { + super(ctx); + this.conn = conn; + this.tableName = tableName; + this.keyOffset = keyOffset; + this.keyRange = keyRange; + this.listLength = listLength; + this.numLists = numLists; + } + + public void doWork() throws SQLException { + ResultSet rs = conn.createStatement().executeQuery( + "SELECT a_prev FROM " + tableName + + " WHERE a_key >= " + Integer.toString(keyOffset) + + " AND a_key < " + Integer.toString(keyOffset + keyRange)); + while (rs.next()) { + int prevKey = rs.getInt(1); + assertTrue(prevKey > (listLength-1)*numLists && prevKey <= listLength*numLists); + + ResultSet nextNodeSeeker; + for(int i = 0; i < listLength-1; ++i){ + int nextKey; + nextNodeSeeker = conn.createStatement().executeQuery( + "SELECT a_prev FROM " + tableName + + " WHERE a_key = " + Integer.toString(prevKey)); + if(nextNodeSeeker.next()){ + nextKey = nextNodeSeeker.getInt(1); + assertTrue(nextKey > ((listLength-2)-i)*numLists && nextKey <= ((listLength-1)-i)*numLists); + prevKey = nextKey; + } else { + fail("There is a hole in a linked list. Key " + Integer.toString(prevKey) + " resulted" + + " in an empty query result."); + } + } + } + } + } + + @Test + public void testContinuousIngest() throws Exception { + Connection conn = DriverManager.getConnection(getUrl()); + String query = TestTable; + + PreparedStatement statement = conn.prepareStatement(query); + statement.execute(); + synchronized (conn){ + conn.commit(); + } + + //START: Test Configurations + int startingOffset = 1; + int listLength = 25; + int numLists = 10000; + int numWriters = 5; + int keysPerWriter = numLists/numWriters; + //END: Test Configurations + + //START: Spawn Writer Threads. + // These prepare the node information in the form of table rows. + MultithreadedTestUtil.TestContext ctx = new MultithreadedTestUtil.TestContext(getTestClusterConfig()); + AtomicInteger sharedPrevKeyCounter = new AtomicInteger((listLength-1)*numLists+1); + for(int i = 0; i c) { + Annotation[] annotations = c.getAnnotations(); + for (Annotation curAnnotation : annotations) { + if (curAnnotation.toString().contains("NeedsOwnMiniClusterTest")) { + /* Skip tests that aren't designed to run against a live cluster. + * For a live cluster, we cannot bring it up and down as required + * for these tests to run. + */ + return false; + } + } + return testFilterRe.matcher(c.getName()).find() && + // Our pattern will match the below NON-IntegrationTest. Rather than + // do exotic regex, just filter it out here + super.isCandidateClass(c); + } + } + + @Override + protected void addOptions() { + addOptWithArg(SHORT_REGEX_ARG, + "Java regex to use selecting tests to run: e.g. .*TestBig.*" + + " will select all tests that include TestBig in their name. Default: " + + ".*end2end.*"); + addOptNoArg(SKIP_TESTS, + "Print list of End2End test suits without running them."); + } + + @Override + protected void processOptions(CommandLine cmd) { + String testFilterString = cmd.getOptionValue(SHORT_REGEX_ARG, null); + if (testFilterString != null) { + end2endTestFilter.setPattern(testFilterString); + } + skipTests = cmd.hasOption(SKIP_TESTS); + } + + /** + * Returns test classes annotated with @Category(IntegrationTests.class), + * according to the filter specific on the command line (if any). + */ + private Class[] findEnd2EndTestClasses() + throws ClassNotFoundException, LinkageError, IOException { + End2EndFileNameFilter nameFilter = new End2EndFileNameFilter(); + ClassFinder classFinder = new ClassFinder(null, nameFilter, end2endTestFilter); + Set> classes = classFinder.findClasses("org.apache.phoenix.end2end", true); + return classes.toArray(new Class[classes.size()]); + } + + + public static class End2EndTestListenter extends TextListener { + private final PrintStream fWriter; + List completes; + public End2EndTestListenter(PrintStream writer) { + super(writer); + completes = new ArrayList(); + fWriter = writer; + } + + @Override + protected void printHeader(long runTime) { + fWriter.println(); + fWriter.println("=========== Test Result ==========="); + fWriter.println("Time: " + elapsedTimeAsString(runTime)); + } + + @Override + public void testStarted(Description description) { + fWriter.println(); + fWriter.println("===> " + description.getDisplayName() + " starts"); + } + + @Override + public void testFinished(Description description) throws Exception { + super.testFinished(description); + completes.add(description.getDisplayName()); + } + + void printSummary(Result result){ + Set failures = new HashSet(); + for(Failure f : result.getFailures()){ + failures.add(f.getTestHeader()); + } + fWriter.println(); + fWriter.println("==== Test Summary ===="); + String status; + for(String curTest : completes){ + status = "passed"; + if(failures.contains(curTest)) { + status = "failed"; + } + fWriter.println(curTest + " " + status + "!"); + } + } + + @Override + public void testRunFinished(Result result) { + printHeader(result.getRunTime()); + printFailures(result); + printSummary(result); + fWriter.println(); + printFooter(result); + } + }; + + + @Override + protected int doWork() throws Exception { + //this is called from the command line, so we should set to use the distributed cluster + IntegrationTestingUtility.setUseDistributedCluster(conf); + System.out.println(System.getProperty("testproperty")); + Class[] classes = findEnd2EndTestClasses(); + System.out.println("Found " + classes.length + " end2end tests to run:"); + for (Class aClass : classes) { + System.out.println(" " + aClass); + } + if(skipTests) return 0; + + JUnitCore junit = new JUnitCore(); + junit.addListener(new End2EndTestListenter(System.out)); + Result result = junit.run(classes); + + return result.wasSuccessful() ? 0 : 1; + } +}