gjacoby126 commented on a change in pull request #457: PHOENIX-5190 Implement
TaskRegionObserver for Index rebuild
URL: https://github.com/apache/phoenix/pull/457#discussion_r268361268
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/TaskRegionObserver.java
##########
@@ -214,85 +165,77 @@ public DropChildViewsTask(RegionCoprocessorEnvironment
env, long timeMaxInterval
@Override
public void run() {
PhoenixConnection connForTask = null;
- Timestamp timestamp = null;
- String tenantId = null;
- byte[] tenantIdBytes;
- String schemaName= null;
- byte[] schemaNameBytes;
- String tableName = null;
- byte[] tableNameBytes;
- PhoenixConnection pconn;
try {
- String taskQuery = "SELECT " +
- PhoenixDatabaseMetaData.TASK_TS + ", " +
- PhoenixDatabaseMetaData.TENANT_ID + ", " +
- PhoenixDatabaseMetaData.TABLE_SCHEM + ", " +
- PhoenixDatabaseMetaData.TABLE_NAME +
- " FROM " + PhoenixDatabaseMetaData.SYSTEM_TASK_NAME +
- " WHERE "+ PhoenixDatabaseMetaData.TASK_TYPE + " = " +
PTable.TaskType.DROP_CHILD_VIEWS.getSerializedValue();
-
connForTask =
QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
- PreparedStatement taskStatement =
connForTask.prepareStatement(taskQuery);
- ResultSet rs = taskStatement.executeQuery();
- while (rs.next()) {
+ String[] excludeStates = new String[] {
PTable.TaskStatus.FAILED.toString(),
+ PTable.TaskStatus.COMPLETED.toString() };
+ List<Task.TaskRecord> taskRecords =
Task.queryTaskTable(connForTask, excludeStates);
+ for (Task.TaskRecord taskRecord : taskRecords){
try {
- // delete child views only if the parent table is
deleted from the system catalog
- timestamp = rs.getTimestamp(1);
- tenantId = rs.getString(2);
- tenantIdBytes= rs.getBytes(2);
- schemaName= rs.getString(3);
- schemaNameBytes = rs.getBytes(3);
- tableName= rs.getString(4);
- tableNameBytes = rs.getBytes(4);
-
- if (tenantId != null) {
- Properties tenantProps = new Properties();
-
tenantProps.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
- pconn =
QueryUtil.getConnectionOnServer(tenantProps,
env.getConfiguration()).unwrap(PhoenixConnection.class);
-
- }
- else {
- pconn =
QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ TaskType taskType = taskRecord.getTaskType();
+ if (!classMap.containsKey(taskType)) {
+ LOG.warn("Don't know how to execute task type: " +
taskType.name());
+ continue;
}
- MetaDataProtocol.MetaDataMutationResult result = new
MetaDataClient(pconn).updateCache(pconn.getTenantId(),
- schemaName, tableName, true);
- if (result.getMutationCode() !=
MetaDataProtocol.MutationCode.TABLE_ALREADY_EXISTS) {
- MetaDataEndpointImpl.dropChildViews(env,
tenantIdBytes, schemaNameBytes, tableNameBytes);
- } else if (System.currentTimeMillis() <
timeMaxInterval + timestamp.getTime()) {
- // skip this task as it has not been expired and
its parent table has not been dropped yet
- LOG.info("Skipping a child view drop task. The
parent table has not been dropped yet : " +
- schemaName + "." + tableName +
- " with tenant id " + (tenantId == null ? "
IS NULL" : tenantId) +
- " and timestamp " + timestamp.toString());
+ String className = classMap.get(taskType);
+
+ Class<?> concreteClass = Class.forName(className);
+
+ Object obj = concreteClass.newInstance();
+ Method runMethod =
concreteClass.getDeclaredMethod("run",
+ Task.TaskRecord.class);
+ Method initMethod =
concreteClass.getSuperclass().getDeclaredMethod("init",
+ RegionCoprocessorEnvironment.class,
Long.class);
+ initMethod.invoke(obj, env, timeMaxInterval);
+
+ // Change task status to STARTED
+ Task.addTask(connForTask, taskRecord.getTaskType(),
taskRecord.getTenantId(), taskRecord.getSchemaName(),
Review comment:
If one server runs TaskRegionObserver, and it runs this line to set a Task
as STARTED, and then immediately afterward the server dies, will the next
server that runs TaskRegionObserver pick up the orphaned task and actually run
it? Since you only screen out FAILED and COMPLETE above, I think the answer's
yes, but making sure.
----------------------------------------------------------------
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
