kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From liy...@apache.org
Subject [1/5] kylin git commit: KYLIN-3194 use BrokenExecutable to tolerate missing job classes
Date Sun, 04 Feb 2018 05:16:15 GMT
Repository: kylin
Updated Branches:
  refs/heads/master c77183343 -> 76dc04971


KYLIN-3194 use BrokenExecutable to tolerate missing job classes


Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/f797c840
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/f797c840
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/f797c840

Branch: refs/heads/master
Commit: f797c84044f52f9c351ad680bc3a8a3482781121
Parents: c771833
Author: Li Yang <liyang@apache.org>
Authored: Wed Jan 24 15:10:07 2018 +0800
Committer: Li Yang <liyang@apache.org>
Committed: Sun Feb 4 08:44:22 2018 +0800

----------------------------------------------------------------------
 .../kylin/job/execution/AbstractExecutable.java |   2 +-
 .../kylin/job/execution/BrokenExecutable.java   |  47 +++++++
 .../kylin/job/execution/ExecutableManager.java  | 124 ++++---------------
 .../apache/kylin/job/ExecutableManagerTest.java |  16 +--
 .../d9a2b721-9916-4607-8047-148ceb2473b1        |  14 +++
 .../apache/kylin/rest/service/JobService.java   |   9 +-
 6 files changed, 102 insertions(+), 110 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/f797c840/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
index 1a84871..91283f0 100644
--- a/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
@@ -259,7 +259,7 @@ public abstract class AbstractExecutable implements Executable, Idempotent
{
     }
 
     @Override
-    public final ExecutableState getStatus() {
+    public ExecutableState getStatus() {
         ExecutableManager manager = getManager();
         return manager.getOutput(this.getId()).getState();
     }

http://git-wip-us.apache.org/repos/asf/kylin/blob/f797c840/core-job/src/main/java/org/apache/kylin/job/execution/BrokenExecutable.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/BrokenExecutable.java b/core-job/src/main/java/org/apache/kylin/job/execution/BrokenExecutable.java
new file mode 100644
index 0000000..7bb4fe2
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/BrokenExecutable.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.job.execution;
+
+import org.apache.kylin.job.exception.ExecuteException;
+
+/**
+ * A special Executable used to indicate any executable whose metadata is broken.
+ */
+public class BrokenExecutable extends AbstractExecutable {
+    
+    public BrokenExecutable() {
+        super();
+    }
+
+    @Override
+    public String getName() {
+        return "[BROKEN] " + super.getName();
+    }
+    
+    @Override
+    public ExecutableState getStatus() {
+        return ExecutableState.DISCARDED;
+    }
+    
+    @Override
+    protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+        throw new UnsupportedOperationException();
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kylin/blob/f797c840/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
index 83582d2..0069c03 100644
--- a/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
@@ -23,7 +23,6 @@ import static org.apache.kylin.job.constant.ExecutableConstants.YARN_APP_ID;
 import static org.apache.kylin.job.constant.ExecutableConstants.YARN_APP_URL;
 
 import java.io.IOException;
-import java.lang.reflect.Constructor;
 import java.util.HashMap;
 import java.util.IllegalFormatException;
 import java.util.List;
@@ -239,45 +238,6 @@ public class ExecutableManager {
         }
     }
 
-    /**
-     * Since ExecutableManager will instantiate all AbstractExecutable class by Class.forName(),
but for each version release,
-     * new classes are introduced, old classes are deprecated, renamed or removed. The Class.forName()
will throw out
-     * ClassNotFoundException. This API is used to retrieve the Executable Object list, not
for calling the object method,
-     * so we could just instance the parent common class instead of the concrete class. It
will tolerate the class missing issue.
-     *
-     * @param timeStartInMillis
-     * @param timeEndInMillis
-     * @param expectedClass
-     * @return
-     */
-    public List<AbstractExecutable> getAllAbstractExecutables(long timeStartInMillis,
long timeEndInMillis,
-            Class<? extends AbstractExecutable> expectedClass) {
-        try {
-            List<AbstractExecutable> ret = Lists.newArrayList();
-            for (ExecutablePO po : executableDao.getJobs(timeStartInMillis, timeEndInMillis))
{
-                try {
-                    AbstractExecutable ae = parseToAbstract(po, expectedClass);
-                    ret.add(ae);
-                } catch (IllegalArgumentException e) {
-                    logger.error("error parsing one executabePO: ", e);
-                }
-            }
-            return ret;
-        } catch (PersistentException e) {
-            logger.error("error get All Jobs", e);
-            throw new RuntimeException(e);
-        }
-    }
-
-    public AbstractExecutable getAbstractExecutable(String uuid, Class<? extends AbstractExecutable>
expectedClass) {
-        try {
-            return parseToAbstract(executableDao.getJob(uuid), expectedClass);
-        } catch (PersistentException e) {
-            logger.error("fail to get job:" + uuid, e);
-            throw new RuntimeException(e);
-        }
-    }
-
     public List<String> getAllJobIds() {
         try {
             return executableDao.getJobIds();
@@ -536,70 +496,40 @@ public class ExecutableManager {
             return null;
         }
         String type = executablePO.getType();
-        try {
-            Class<? extends AbstractExecutable> clazz = ClassUtil.forName(type, AbstractExecutable.class);
-            Constructor<? extends AbstractExecutable> constructor = clazz.getConstructor();
-            AbstractExecutable result = constructor.newInstance();
-            result.initConfig(config);
-            result.setId(executablePO.getUuid());
-            result.setName(executablePO.getName());
-            result.setParams(executablePO.getParams());
-            List<ExecutablePO> tasks = executablePO.getTasks();
-            if (tasks != null && !tasks.isEmpty()) {
-                Preconditions.checkArgument(result instanceof ChainedExecutable);
-                for (ExecutablePO subTask : tasks) {
-                    ((ChainedExecutable) result).addTask(parseTo(subTask));
-                }
+        AbstractExecutable result = newExecutable(type);
+        result.initConfig(config);
+        result.setId(executablePO.getUuid());
+        result.setName(executablePO.getName());
+        result.setParams(executablePO.getParams());
+        List<ExecutablePO> tasks = executablePO.getTasks();
+        if (tasks != null && !tasks.isEmpty()) {
+            Preconditions.checkArgument(result instanceof ChainedExecutable);
+            for (ExecutablePO subTask : tasks) {
+                ((ChainedExecutable) result).addTask(parseTo(subTask));
             }
-            List<ExecutablePO> tasksForCheck = executablePO.getTasksForCheck();
-            if (tasksForCheck != null && !tasksForCheck.isEmpty()) {
-                Preconditions.checkArgument(result instanceof CheckpointExecutable);
-                for (ExecutablePO subTaskForCheck : tasksForCheck) {
-                    ((CheckpointExecutable) result).addTaskForCheck(parseTo(subTaskForCheck));
-                }
+        }
+        List<ExecutablePO> tasksForCheck = executablePO.getTasksForCheck();
+        if (tasksForCheck != null && !tasksForCheck.isEmpty()) {
+            Preconditions.checkArgument(result instanceof CheckpointExecutable);
+            for (ExecutablePO subTaskForCheck : tasksForCheck) {
+                ((CheckpointExecutable) result).addTaskForCheck(parseTo(subTaskForCheck));
             }
-            return result;
-        } catch (ReflectiveOperationException e) {
-            throw new IllegalStateException("cannot parse this job:" + executablePO.getId(),
e);
         }
+        return result;
     }
 
-    private AbstractExecutable parseToAbstract(ExecutablePO executablePO,
-            Class<? extends AbstractExecutable> expectedClass) {
-        if (executablePO == null) {
-            logger.warn("executablePO is null");
-            return null;
+    private AbstractExecutable newExecutable(String type) {
+        Class<? extends AbstractExecutable> clazz;
+        try {
+            clazz = ClassUtil.forName(type, AbstractExecutable.class);
+        } catch (ClassNotFoundException ex) {
+            clazz = BrokenExecutable.class;
+            logger.error("Unknown executable type '" + type + "', using BrokenExecutable");
         }
-        String type = executablePO.getType();
         try {
-            Class<? extends AbstractExecutable> clazz = null;
-            try {
-                clazz = ClassUtil.forName(type, AbstractExecutable.class);
-            } catch (ClassNotFoundException e) {
-                clazz = ClassUtil.forName(expectedClass.getName(), AbstractExecutable.class);
-            }
-            Constructor<? extends AbstractExecutable> constructor = clazz.getConstructor();
-            AbstractExecutable result = constructor.newInstance();
-            result.initConfig(config);
-            result.setId(executablePO.getUuid());
-            result.setName(executablePO.getName());
-            result.setParams(executablePO.getParams());
-            List<ExecutablePO> tasks = executablePO.getTasks();
-            if (tasks != null && !tasks.isEmpty()) {
-                Preconditions.checkArgument(result instanceof ChainedExecutable);
-                for (ExecutablePO subTask : tasks) {
-                    AbstractExecutable parseToTask = null;
-                    try {
-                        parseToTask = parseTo(subTask);
-                    } catch (IllegalStateException e) {
-                        parseToTask = parseToAbstract(subTask, DefaultChainedExecutable.class);
-                    }
-                    ((ChainedExecutable) result).addTask(parseToTask);
-                }
-            }
-            return result;
-        } catch (ReflectiveOperationException e) {
-            throw new IllegalStateException("cannot parse this job:" + executablePO.getId(),
e);
+            return clazz.getConstructor().newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to instantiate " + clazz, e);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/f797c840/core-job/src/test/java/org/apache/kylin/job/ExecutableManagerTest.java
----------------------------------------------------------------------
diff --git a/core-job/src/test/java/org/apache/kylin/job/ExecutableManagerTest.java b/core-job/src/test/java/org/apache/kylin/job/ExecutableManagerTest.java
index faea9a4..73f0410 100644
--- a/core-job/src/test/java/org/apache/kylin/job/ExecutableManagerTest.java
+++ b/core-job/src/test/java/org/apache/kylin/job/ExecutableManagerTest.java
@@ -47,12 +47,6 @@ public class ExecutableManagerTest extends LocalFileMetadataTestCase {
     public void setup() throws Exception {
         createTestMetadata();
         service = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
-
-        for (String jobId : service.getAllJobIds()) {
-            System.out.println("deleting " + jobId);
-            service.deleteJob(jobId);
-        }
-
     }
 
     @After
@@ -63,13 +57,21 @@ public class ExecutableManagerTest extends LocalFileMetadataTestCase {
     @Test
     public void test() throws Exception {
         assertNotNull(service);
+        
+        // all existing are broken jobs
+        List<AbstractExecutable> existing = service.getAllExecutables();
+        for (AbstractExecutable exec : existing) {
+            assertEquals("BrokenExecutable", exec.getClass().getSimpleName());
+            assertEquals(ExecutableState.DISCARDED, exec.getStatus());
+        }
+        
         BaseTestExecutable executable = new SucceedTestExecutable();
         executable.setParam("test1", "test1");
         executable.setParam("test2", "test2");
         executable.setParam("test3", "test3");
         service.addJob(executable);
         List<AbstractExecutable> result = service.getAllExecutables();
-        assertEquals(1, result.size());
+        assertEquals(existing.size() + 1, result.size());
         AbstractExecutable another = service.getJob(executable.getId());
         assertJobEqual(executable, another);
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/f797c840/examples/test_case_data/localmeta/execute/d9a2b721-9916-4607-8047-148ceb2473b1
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/execute/d9a2b721-9916-4607-8047-148ceb2473b1
b/examples/test_case_data/localmeta/execute/d9a2b721-9916-4607-8047-148ceb2473b1
new file mode 100644
index 0000000..fb8ff30
--- /dev/null
+++ b/examples/test_case_data/localmeta/execute/d9a2b721-9916-4607-8047-148ceb2473b1
@@ -0,0 +1,14 @@
+{
+  "uuid" : "d9a2b721-9916-4607-8047-148ceb2473b1",
+  "last_modified" : 1516778161249,
+  "version" : "2.3.0",
+  "name" : null,
+  "tasks" : null,
+  "tasks_check" : null,
+  "type" : "org.apache.kylin.job.BadClassName",
+  "params" : {
+    "test2" : "test2",
+    "test3" : "test3",
+    "test1" : "test1"
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kylin/blob/f797c840/server-base/src/main/java/org/apache/kylin/rest/service/JobService.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/JobService.java b/server-base/src/main/java/org/apache/kylin/rest/service/JobService.java
index e447031..5f80b84 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/service/JobService.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/JobService.java
@@ -725,9 +725,9 @@ public class JobService extends BasicService implements InitializingBean
{
     public List<CubingJob> innerSearchCubingJobs(final String cubeName, final String
jobName,
             final Set<ExecutableState> statusList, long timeStartInMillis, long timeEndInMillis,
             final Map<String, Output> allOutputs, final boolean nameExactMatch, final
String projectName) {
-        List<CubingJob> results = Lists.newArrayList(FluentIterable.from(
-                getExecutableManager().getAllAbstractExecutables(timeStartInMillis, timeEndInMillis,
CubingJob.class))
-                .filter(new Predicate<AbstractExecutable>() {
+        List<CubingJob> results = Lists.newArrayList(
+                FluentIterable.from(getExecutableManager().getAllExecutables(timeStartInMillis,
timeEndInMillis))
+                        .filter(new Predicate<AbstractExecutable>() {
                     @Override
                     public boolean apply(AbstractExecutable executable) {
                         if (executable instanceof CubingJob) {
@@ -824,8 +824,7 @@ public class JobService extends BasicService implements InitializingBean
{
         List<CheckpointExecutable> results = Lists
                 .newArrayList(
                         FluentIterable
-                                .from(getExecutableManager().getAllAbstractExecutables(timeStartInMillis,
-                                        timeEndInMillis, CheckpointExecutable.class))
+                                .from(getExecutableManager().getAllExecutables(timeStartInMillis,
timeEndInMillis))
                                 .filter(new Predicate<AbstractExecutable>() {
                                     @Override
                                     public boolean apply(AbstractExecutable executable) {


Mime
View raw message