phoenix-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] [phoenix] gjacoby126 commented on a change in pull request #457: PHOENIX-5190 Implement TaskRegionObserver for Index rebuild
Date Fri, 22 Mar 2019 23:05:05 GMT
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:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message