[kylin] branch master updated: KYLIN-2431 Use StorageCleanupJob cleanup intermediate hive tables after check whether it was created by current deployment

[nic]

This is an automated email from the ASF dual-hosted git repository.

nic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kylin.git


The following commit(s) were added to refs/heads/master by this push:
     new 7de98cf  KYLIN-2431 Use StorageCleanupJob cleanup intermediate hive 
tables after check whether it was created by current deployment
7de98cf is described below

commit 7de98cfceed810f1bd7d4c058173ddb3f4eff3ee
Author: yaqian.zhang <598593...@qq.com>
AuthorDate: Tue Oct 15 16:48:01 2019 +0800

    KYLIN-2431 Use StorageCleanupJob cleanup intermediate hive tables after 
check whether it was created by current deployment
---
 .../apache/kylin/rest/job/StorageCleanupJob.java   | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git 
a/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java 
b/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java
index a396e9d..491107f 100755
--- a/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java
@@ -325,6 +325,7 @@ public class StorageCleanupJob extends AbstractApplication {
 
         List<String> allJobs = executableManager.getAllJobIds();
         List<String> workingJobList = new ArrayList<String>();
+        List<String> allUuids = getAllUuids(allJobs);
         Map<String, String> segmentId2JobId = Maps.newHashMap();
 
         for (String jobId : allJobs) {
@@ -376,6 +377,11 @@ public class StorageCleanupJob extends AbstractApplication 
{
                 continue;
             }
 
+            if (!allUuids.contains(uuid)) {
+                logger.debug("Skip table because is not current deployment 
create, " + tableName);
+                continue;
+            }
+
             //Some intermediate table ends with job's uuid
             if (allJobs.contains(uuid)) {
                 isNeedDel = !workingJobList.contains(uuid);
@@ -462,6 +468,22 @@ public class StorageCleanupJob extends AbstractApplication 
{
         }
     }
 
+    private List<String> getAllUuids(List<String> allJobs) {
+        List<String> allUuids = new ArrayList<>();
+        for (String jobId : allJobs) {
+            allUuids.add(jobId);
+            try {
+                String segmentId = getSegmentIdFromJobId(jobId);
+                if (segmentId != null) {
+                    allUuids.add(segmentId);
+                }
+            } catch (Exception ex) {
+                logger.warn("Failed to find segment ID from job ID " + jobId + 
", ignore it");
+            }
+        }
+        return allUuids;
+    }
+
     private String getSegmentIdFromJobId(String jobId) {
         AbstractExecutable abstractExecutable = 
executableManager.getJob(jobId);
         String segmentId = abstractExecutable.getParam("segmentId");