PHOENIX-6329 Eliminate or serialize minicluster restart operations in…

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);
  }
 }

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,13 +104,19 @@ 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;
 
  @Captor
  private ArgumentCaptor<String []> 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,

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,18 +170,18 @@
  * 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.
  * 
  * 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<Void>() {
- @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<Void> 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<HTableDescriptor> tableDescriptors = Arrays.asList(admin.listTables());
+ int tableCount = tableDescriptors.size();
+
+ int retryCount=10;
+ List<Future<Void>> futures = new ArrayList<>();
+ while (!tableDescriptors.isEmpty() && retryCount-->0) {
+ for(final HTableDescriptor tableDescriptor : tableDescriptors) {
+ futures.add(dropHTableExecutor.submit(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ if (admin.isTableEnabled(tableDescriptor.getTableName())) {
+ admin.disableTable(tableDescriptor.getTableName());
+ }
+ admin.deleteTable(tableDescriptor.getTableName());
+ return null;
+ }
+ }));
+ }
+ for (Future<Void> 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<List<Object>>... expectedResultsArray) throws SQLException {
  List<List<Object>> 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);

pom.xml

@@ -68,7 +68,7 @@
  <parent>
  <groupId>org.apache</groupId>
  <artifactId>apache</artifactId>
- <version>21</version>
+ <version>23</version>
  </parent>
 
  <scm>
@@ -145,6 +145,8 @@
  <restrict-imports.enforcer.version>1.1.0</restrict-imports.enforcer.version>
  <maven-shade-plugin.version>3.2.4</maven-shade-plugin.version>
  <maven-project-info-reports-plugin.version>3.1.1</maven-project-info-reports-plugin.version>
+ <!-- Override property in ASF parent -->
+ <surefire.version>2.22.2</surefire.version>
  <spotbugs-maven-plugin.version>4.1.3</spotbugs-maven-plugin.version>
  <spotbugs.version>4.1.3</spotbugs.version>
  <jacoco-maven-plugin.version>0.8.5</jacoco-maven-plugin.version>