diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java index 15d0e4964df..d6eff3a31e6 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java @@ -219,6 +219,6 @@ public void tickTime() { @AfterClass public static synchronized void teardown() { - tearDownMiniClusterAsync(2); + tearDownMiniCluster(2); } } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java index 9ae4426d0f6..7c53701a751 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java @@ -39,6 +39,7 @@ import org.apache.phoenix.util.SchemaUtil; import org.junit.Assert; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; @@ -103,6 +104,7 @@ public class ParameterizedIndexUpgradeToolIT extends BaseTest { private Connection connTenant; private Admin admin; private IndexUpgradeTool iut; + private static String tmpDir = System.getProperty("java.io.tmpdir");; @Mock private IndexTool indexToolMock; @@ -110,6 +112,11 @@ public class ParameterizedIndexUpgradeToolIT extends BaseTest { @Captor private ArgumentCaptor argCapture; + @BeforeClass + public static synchronized void saveTmp () throws Exception { + tmpDir = System.getProperty("java.io.tmpdir"); + } + @Before public void setup () throws Exception { MockitoAnnotations.initMocks(this); @@ -144,7 +151,8 @@ private void setClusterProperties() { .get(QueryServices.INDEX_REGION_OBSERVER_ENABLED_ATTRIB)) || Boolean.toString(!isNamespaceEnabled).equals(serverProps .get(QueryServices.IS_NAMESPACE_MAPPING_ENABLED))) { - tearDownMiniClusterAsync(1); + tearDownMiniCluster(1); + System.setProperty("java.io.tmpdir", tmpDir); } //setting up properties for namespace clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, diff --git a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java index 1537ce65b78..bd3025e8025 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java @@ -125,6 +125,7 @@ import org.apache.hadoop.hbase.MiniHBaseCluster; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.HBaseAdmin; import org.apache.hadoop.hbase.ipc.PhoenixRpcSchedulerFactory; import org.apache.hadoop.hbase.master.AssignmentManager; @@ -169,7 +170,7 @@ * Base class that contains all the methods needed by * client-time and hbase-time managed tests. * - * * Tests using a mini cluster need to be classified either + * Tests using a mini cluster need to be classified either * as {@link ParallelStatsDisabledTest} or {@link ParallelStatsEnabledTest} * or {@link NeedsOwnMiniClusterTest} otherwise they won't be run * when one runs mvn verify or mvn install. @@ -177,10 +178,10 @@ * For tests needing connectivity to a cluster, please use * {@link ParallelStatsDisabledIt} or {@link ParallelStatsEnabledIt}. * - * In the case when a test can't share the same mini cluster as the + * In the case when a test can't share the same mini cluster as the * ones used by {@link ParallelStatsDisabledIt} or {@link ParallelStatsEnabledIt}, * one could extend this class and spin up your own mini cluster. Please - * make sure to annotate such clesses with {@link NeedsOwnMiniClusterTest} and + * make sure to annotate such classes with {@link NeedsOwnMiniClusterTest} and * shutdown the mini cluster in a method annotated by @AfterClass. * */ @@ -416,21 +417,6 @@ protected static String getZKClientPort(Configuration conf) { protected static HBaseTestingUtility utility; protected static final Configuration config = HBaseConfiguration.create(); - private static class TearDownMiniClusterThreadFactory implements ThreadFactory { - private static final AtomicInteger threadNumber = new AtomicInteger(1); - private static final String NAME_PREFIX = "PHOENIX-TEARDOWN-MINICLUSTER-thread-"; - - @Override - public Thread newThread(Runnable r) { - Thread t = new Thread(r, NAME_PREFIX + threadNumber.getAndIncrement()); - t.setDaemon(true); - return t; - } - } - - private static ExecutorService tearDownClusterService = - Executors.newSingleThreadExecutor(new TearDownMiniClusterThreadFactory()); - protected static String getUrl() { if (!clusterInitialized) { throw new IllegalStateException("Cluster must be initialized before attempting to get the URL"); @@ -480,51 +466,47 @@ protected static void dropNonSystemTables() throws Exception { } } - public static void tearDownMiniClusterAsync(final int numTables) { - final HBaseTestingUtility u = utility; + //Note that newer miniCluster versions will overwrite "java.io.tmpdir" system property. + //After you shut down the minicluster, it will point to a non-existent directory + //You will need to save the original "java.io.tmpdir" before starting the miniCluster, and + //restore it after shutting it down, if you want to keep using the JVM. + public static synchronized void tearDownMiniCluster(final int numTables) { + long startTime = System.currentTimeMillis(); try { + ConnectionFactory.shutdown(); destroyDriver(); - utility = null; - clusterInitialized = false; + utility.shutdownMiniMapReduceCluster(); + } catch (Throwable t) { + LOGGER.error("Exception caught when shutting down mini map reduce cluster", t); } finally { - tearDownClusterService.submit(new Callable() { - @Override - public Void call() throws Exception { - long startTime = System.currentTimeMillis(); - if (u != null) { - try { - u.shutdownMiniMapReduceCluster(); - } catch (Throwable t) { - LOGGER.error( - "Exception caught when shutting down mini map reduce cluster", t); - } finally { - try { - u.shutdownMiniCluster(); - } catch (Throwable t) { - LOGGER.error("Exception caught when shutting down mini cluster", t); - } finally { - try { - ConnectionFactory.shutdown(); - } finally { - LOGGER.info( - "Time in seconds spent in shutting down mini cluster with " - + numTables + " tables: " - + (System.currentTimeMillis() - startTime) / 1000); - } - } - } - } - return null; - } - }); + try { + utility.shutdownMiniCluster(); + } catch (Throwable t) { + LOGGER.error("Exception caught when shutting down mini cluster", t); + } finally { + clusterInitialized = false; + utility = null; + LOGGER.info("Time in seconds spent in shutting down mini cluster with " + numTables + + " tables: " + (System.currentTimeMillis() - startTime) / 1000); + } + } + } + + public static synchronized void resetHbase() { + try { + ConnectionFactory.shutdown(); + destroyDriver(); + disableAndDropAllTables(); + } catch (Exception e) { + LOGGER.error("Error resetting HBase"); } } - protected static void setUpTestDriver(ReadOnlyProps props) throws Exception { + protected static synchronized void setUpTestDriver(ReadOnlyProps props) throws Exception { setUpTestDriver(props, props); } - protected static void setUpTestDriver(ReadOnlyProps serverProps, ReadOnlyProps clientProps) throws Exception { + protected static synchronized void setUpTestDriver(ReadOnlyProps serverProps, ReadOnlyProps clientProps) throws Exception { if (driver == null) { String url = checkClusterInitialized(serverProps); driver = initAndRegisterTestDriver(url, clientProps); @@ -548,7 +530,7 @@ private static boolean isDistributedClusterModeEnabled(Configuration conf) { * @return url to be used by clients to connect to the mini cluster. * @throws Exception */ - private static String initMiniCluster(Configuration conf, ReadOnlyProps overrideProps) throws Exception { + private static synchronized String initMiniCluster(Configuration conf, ReadOnlyProps overrideProps) throws Exception { setUpConfigForMiniCluster(conf, overrideProps); utility = new HBaseTestingUtility(conf); try { @@ -634,8 +616,6 @@ public static Configuration setUpConfigForMiniCluster(Configuration conf, ReadOn conf.setInt("hbase.assignment.zkevent.workers", 5); conf.setInt("hbase.assignment.threads.max", 5); conf.setInt("hbase.catalogjanitor.interval", 5000); - //Allow for an extra long miniCluster startup time in case of an overloaded test machine - conf.setInt("hbase.master.start.timeout.localHBaseCluster", 200000); conf.setInt(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB, 10000); conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2); conf.setInt(NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY, 1); @@ -670,7 +650,7 @@ private static PhoenixTestDriver newTestDriver(ReadOnlyProps props) throws Excep * Create a {@link PhoenixTestDriver} and register it. * @return an initialized and registered {@link PhoenixTestDriver} */ - public static PhoenixTestDriver initAndRegisterTestDriver(String url, ReadOnlyProps props) throws Exception { + public static synchronized PhoenixTestDriver initAndRegisterTestDriver(String url, ReadOnlyProps props) throws Exception { PhoenixTestDriver newDriver = newTestDriver(props); DriverManager.registerDriver(newDriver); Driver oldDriver = DriverManager.getDriver(url); @@ -684,7 +664,7 @@ public static PhoenixTestDriver initAndRegisterTestDriver(String url, ReadOnlyPr } //Close and unregister the driver. - protected static boolean destroyDriver(Driver driver) { + protected static synchronized boolean destroyDriver(Driver driver) { if (driver != null) { assert(driver instanceof PhoenixEmbeddedDriver); PhoenixEmbeddedDriver pdriver = (PhoenixEmbeddedDriver)driver; @@ -708,7 +688,7 @@ protected static String getOrganizationId() { private static long timestamp; - public static long nextTimestamp() { + public static synchronized long nextTimestamp() { timestamp += 100; return timestamp; } @@ -795,7 +775,7 @@ public static String generateUniqueSequenceName() { return "S" + Integer.toString(MAX_SEQ_SUFFIX_VALUE + nextName).substring(1); } - public static void freeResourcesIfBeyondThreshold() throws Exception { + public static synchronized void freeResourcesIfBeyondThreshold() throws Exception { if (TABLE_COUNTER.get() > TEARDOWN_THRESHOLD) { int numTables = TABLE_COUNTER.get(); TABLE_COUNTER.set(0); @@ -806,9 +786,9 @@ public static void freeResourcesIfBeyondThreshold() throws Exception { deletePriorMetaData(HConstants.LATEST_TIMESTAMP, url); } else { LOGGER.info( - "Shutting down mini cluster because number of tables on this mini cluster is likely greater than " + "Clearing all HBase tables in minicluster because number of tables on this mini cluster is likely greater than " + TEARDOWN_THRESHOLD); - tearDownMiniClusterAsync(numTables); + resetHbase(); } } } @@ -911,7 +891,7 @@ private static void deletePriorSchemas(long ts, String url) throws Exception { } } - protected static void deletePriorMetaData(long ts, String url) throws Exception { + protected static synchronized void deletePriorMetaData(long ts, String url) throws Exception { deletePriorTables(ts, url); if (ts != HConstants.LATEST_TIMESTAMP) { ts = nextTimestamp() - 1; @@ -1574,7 +1554,7 @@ protected static String initSaltedEntityHistoryTableValues(String tableName, Str /** * Disable and drop all non system tables */ - protected static void disableAndDropNonSystemTables() throws Exception { + protected static synchronized void disableAndDropNonSystemTables() throws Exception { if (driver == null) return; HBaseAdmin admin = driver.getConnectionQueryServices(null, null).getAdmin(); try { @@ -1590,7 +1570,8 @@ protected static void disableAndDropNonSystemTables() throws Exception { } } - private static void disableAndDropTable(final HBaseAdmin admin, final TableName tableName) + + private static synchronized void disableAndDropTable(final Admin admin, final TableName tableName) throws Exception { Future future = null; boolean success = false; @@ -1622,6 +1603,49 @@ public Void call() throws Exception { } } } + + private static synchronized void disableAndDropAllTables() throws IOException { + long startTime = System.currentTimeMillis(); + final Admin admin = utility.getHBaseAdmin(); + ExecutorService dropHTableExecutor = Executors + .newCachedThreadPool(factory); + + List tableDescriptors = Arrays.asList(admin.listTables()); + int tableCount = tableDescriptors.size(); + + int retryCount=10; + List> futures = new ArrayList<>(); + while (!tableDescriptors.isEmpty() && retryCount-->0) { + for(final HTableDescriptor tableDescriptor : tableDescriptors) { + futures.add(dropHTableExecutor.submit(new Callable() { + @Override + public Void call() throws Exception { + if (admin.isTableEnabled(tableDescriptor.getTableName())) { + admin.disableTable(tableDescriptor.getTableName()); + } + admin.deleteTable(tableDescriptor.getTableName()); + return null; + } + })); + } + for (Future future : futures) { + try { + future.get(dropTableTimeout, TimeUnit.SECONDS); + } catch (Exception e) { + LOGGER.warn("Error while dropping table, will try again", e); + } + } + tableDescriptors = Arrays.asList(admin.listTables()); + } + if(!tableDescriptors.isEmpty()) { + LOGGER.error("Could not clean up tables!"); + } + dropHTableExecutor.shutdownNow(); + long endTime = System.currentTimeMillis(); + + LOGGER.info("Disabled and dropped {} tables in {} ms", tableCount, endTime-startTime); + + } public static void assertOneOfValuesEqualsResultSet(ResultSet rs, List>... expectedResultsArray) throws SQLException { List> results = Lists.newArrayList(); @@ -1716,7 +1740,7 @@ public static void upsertRow(Connection conn, String fullTableName, int index, b } // Populate the test table with data. - public static void populateTestTable(String fullTableName) throws SQLException { + public static synchronized void populateTestTable(String fullTableName) throws SQLException { Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); try (Connection conn = DriverManager.getConnection(getUrl(), props)) { upsertRows(conn, fullTableName, 3); diff --git a/pom.xml b/pom.xml index 93e7a76226c..ec5b0117a1c 100644 --- a/pom.xml +++ b/pom.xml @@ -68,7 +68,7 @@ org.apache apache - 21 + 23 @@ -145,6 +145,8 @@ 1.1.0 3.2.4 3.1.1 + + 2.22.2 4.1.3 4.1.3 0.8.5