[kylin] 09/12: KYLIN-2431 Use StorageCleanupJob cleanup intermediate hive tables after check whether it was created by current deployment

[ni...@apache.org]

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

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

commit ed8255130860cb4dedcb14842bbca5c8b985bd75
Author: yaqian.zhang <59...@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 c8e73de..9b801fe 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
@@ -305,6 +305,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) {
@@ -356,6 +357,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);
@@ -439,6 +445,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");