swaroopak commented on a change in pull request #660: PHOENIX-5644:
IndexUpgradeTool should sleep only once if there is at …
URL: https://github.com/apache/phoenix/pull/660#discussion_r363547849
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexUpgradeTool.java
##########
@@ -318,53 +323,155 @@ public int executeTool() {
return -1;
}
- private int executeTool(Connection conn, ConnectionQueryServices
queryServices,
+ private int executeTool(Connection conn,
+ ConnectionQueryServices queryServices,
Configuration conf) {
- LOGGER.info("Executing " + operation);
+ ArrayList<String> immutableList = new ArrayList<>();
+ ArrayList<String> mutableList = new ArrayList<>();
for (Map.Entry<String, HashSet<String>> entry
:tablesAndIndexes.entrySet()) {
String dataTableFullName = entry.getKey();
- HashSet<String> indexes = entry.getValue();
-
- try (Admin admin = queryServices.getAdmin()) {
+ try {
PTable dataTable = PhoenixRuntime.getTableNoCache(conn,
dataTableFullName);
- LOGGER.info("Executing " + operation + " of " +
dataTableFullName);
+ if (dataTable.isImmutableRows()) {
+ //add to list where immutable tables are processed in a
different function
+ immutableList.add(dataTableFullName);
+ } else {
+ mutableList.add(dataTableFullName);
+ }
+ } catch (SQLException e) {
+ LOGGER.severe("Something went wrong while getting the PTable "
+ + dataTableFullName + " "+e);
+ return -1;
+ }
+ }
+ long startWaitTime = executeToolForImmutableTables(queryServices,
immutableList);
+ executeToolForMutableTables(conn, queryServices, conf, mutableList);
+ enableImmutableTables(queryServices, immutableList, startWaitTime);
+ rebuildIndexes(conn, conf, immutableList);
+ return 0;
+ }
+ private long executeToolForImmutableTables(ConnectionQueryServices
queryServices,
+ ArrayList<String> immutableList) {
+ LOGGER.info("Started " + operation + " for immutable tables");
+ try (Admin admin = queryServices.getAdmin()) {
+ for (String dataTableFullName : immutableList) {
+ dataTableFullNameGlobal = dataTableFullName;
+ HashSet<String> indexes =
tablesAndIndexes.get(dataTableFullName);
+ LOGGER.info("Executing " + operation + " of " +
dataTableFullName
+ + " (immutable)");
+ disableTable(admin, dataTableFullName, indexes);
+ modifyTable(admin, dataTableFullName, indexes);
+ }
+ } catch (IOException | SQLException e) {
+ LOGGER.severe("Something went wrong while disabling "
+ + "or modifying immutable table " + e);
+ handleFailure(queryServices, dataTableFullNameGlobal,
immutableList);
+ }
+ long startWaitTime = System.currentTimeMillis();
+ return startWaitTime;
+ }
+
+ private void handleFailure(ConnectionQueryServices queryServices,
+ String dataTableFullName,
+ ArrayList<String> tableList) {
+ LOGGER.info("Performing error handling to revert the steps taken
during " +operation);
+ HashSet<String> indexes = tablesAndIndexes.get(dataTableFullName);
+ try (Admin admin = queryServices.getAdmin()) {
+ upgrade = !upgrade;
+ disableTable(admin, dataTableFullName, indexes);
+ modifyTable(admin, dataTableFullName, indexes);
+ enableTable(admin, dataTableFullName, indexes);
+ upgrade = !upgrade;
+ tablesAndIndexes.remove(dataTableFullName); //removing from the map
+ tableList.remove(dataTableFullName); //removing from the list
+ LOGGER.severe(dataTableFullName+" has been removed from the list
as tool failed"
+ + " perform "+operation);
+ } catch (IOException | SQLException e) {
+ LOGGER.severe("Revert of the "+operation +" failed in error
handling, "
+ + "throwing runtime exception");
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void executeToolForMutableTables(Connection conn,
+ ConnectionQueryServices queryServices,
+ Configuration conf,
+ ArrayList<String> mutableTables) {
+ LOGGER.info("Started " + operation + " for mutable tables");
+ try (Admin admin = queryServices.getAdmin()) {
+ for (String dataTableFullName : mutableTables) {
+ dataTableFullNameGlobal = dataTableFullName;
+ HashSet<String> indexes =
tablesAndIndexes.get(dataTableFullName);
+ LOGGER.info("Executing " + operation + " of " +
dataTableFullName);
disableTable(admin, dataTableFullName, indexes);
modifyTable(admin, dataTableFullName, indexes);
- if (dataTable.isImmutableRows()) {
- // If the table is immutable, we need to wait for clients
to purge
- // their caches of table metadata
- LOGGER.info(dataTableFullName + " is an immutable table,
waiting for "
- + (GLOBAL_INDEX_CHECKER_ENABLED_MAP_EXPIRATION_MIN
+ 1)
- + " minutes for client cache to expire");
- if (!(test || dryRun || indexes.isEmpty())) {
- try {
-
Thread.sleep((GLOBAL_INDEX_CHECKER_ENABLED_MAP_EXPIRATION_MIN + 1)
- * 60 * 1000);
- } catch (InterruptedException e) {
- LOGGER.warning("Sleep before starting index
rebuild interrupted. "
- + e.getMessage());
- }
- }
- }
enableTable(admin, dataTableFullName, indexes);
LOGGER.info("Completed " + operation + " of " +
dataTableFullName);
-
- } catch (IOException | SQLException e) {
- LOGGER.severe("Something went wrong while executing " +
operation
- + " steps " + e);
- return -1;
}
+ } catch (IOException | SQLException e) {
+ LOGGER.severe("Something went wrong while executing "
+ + operation + " steps for "+ dataTableFullNameGlobal + " "
+ e);
+ handleFailure(queryServices, dataTableFullNameGlobal,
mutableTables);
}
// Opportunistically kick-off index rebuilds after upgrade operation
- if (upgrade) {
- for (String dataTableFullName : tablesAndIndexes.keySet()) {
- rebuildIndexes(conn, conf, dataTableFullName);
- LOGGER.info("Started index rebuild post " + operation + " of "
- + dataTableFullName);
+ rebuildIndexes(conn, conf, mutableTables);
+ }
+
+ private void enableImmutableTables(ConnectionQueryServices queryServices,
+ ArrayList<String> immutableList,
+ long startWaitTime) {
+ int retryCount = 0;
+ long endWaitTime = System.currentTimeMillis();
+ long waitMore = getWaitMoreTime(endWaitTime, startWaitTime);
+ while (retryCount < 5 && !waited) {
+ // If the table is immutable, we need to wait for clients to purge
+ // their caches of table metadata
+
+ LOGGER.info("waiting for more " + waitMore + " ms for client cache
"
+ + "to expire for immutable tables");
+ try {
+ startWaitTime = System.currentTimeMillis();
+ Thread.sleep(waitMore);
+ waited = true;
+ } catch (InterruptedException e) {
+ endWaitTime = System.currentTimeMillis();
+ waitMore = getWaitMoreTime(endWaitTime, startWaitTime);
+ retryCount++;
+ LOGGER.warning("Sleep before starting index rebuild is
interrupted. "
+ + "Attempting to sleep again! " + e.getMessage());
}
}
- return 0;
+ if (retryCount == 5) {
+ LOGGER.info("Sleep was interrupted multiple times for immutable
tables, "
Review comment:
We want to retry in case of interruption during sleep. And only try a finite
number of times. I think for this corner case scenario, we are okay if we see
any problem with cache incoherency. What say?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
