You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2016/01/22 04:19:42 UTC
[1/2] kylin git commit: Revert "KYLIN-1356 use exec-maven-plugin for
IT environment provision"
Repository: kylin
Updated Branches:
refs/heads/2.0-rc a118a27c6 -> 827a995da
Revert "KYLIN-1356 use exec-maven-plugin for IT environment provision"
This reverts commit ab853f9dc987549ac27dd2bb3db1732e3e474281.
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/827a995d
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/827a995d
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/827a995d
Branch: refs/heads/2.0-rc
Commit: 827a995dab8ed28907d1e16abc5972ae253d712a
Parents: db87d46
Author: honma <ho...@ebay.com>
Authored: Fri Jan 22 11:18:21 2016 +0800
Committer: honma <ho...@ebay.com>
Committed: Fri Jan 22 11:19:33 2016 +0800
----------------------------------------------------------------------
.../engine/spark/BuildCubeWithSparkTest.java | 3 -
.../apache/kylin/job/BuildCubeWithEngine.java | 333 -------------------
.../kylin/job/BuildCubeWithEngineTest.java | 333 +++++++++++++++++++
.../apache/kylin/job/BuildCubeWithStream.java | 110 ------
.../kylin/job/BuildCubeWithStreamTest.java | 114 +++++++
.../apache/kylin/job/BuildIIWithEngineTest.java | 7 +-
.../org/apache/kylin/job/BuildIIWithStream.java | 292 ----------------
.../apache/kylin/job/BuildIIWithStreamTest.java | 290 ++++++++++++++++
dev-support/test_all_against_hdp_2_2_4_2_2.sh | 3 +
pom.xml | 73 ++--
10 files changed, 769 insertions(+), 789 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/engine/spark/BuildCubeWithSparkTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/engine/spark/BuildCubeWithSparkTest.java b/assembly/src/test/java/org/apache/kylin/engine/spark/BuildCubeWithSparkTest.java
index f87fa99..f5b9741 100644
--- a/assembly/src/test/java/org/apache/kylin/engine/spark/BuildCubeWithSparkTest.java
+++ b/assembly/src/test/java/org/apache/kylin/engine/spark/BuildCubeWithSparkTest.java
@@ -46,13 +46,10 @@ import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import com.google.common.collect.Lists;
-//TODO: convert it to a normal class rather than a test case, like in BuildCubeWithEngine
-@Ignore
public class BuildCubeWithSparkTest {
private CubeManager cubeManager;
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngine.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngine.java b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngine.java
deleted file mode 100644
index 2562a8b..0000000
--- a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngine.java
+++ /dev/null
@@ -1,333 +0,0 @@
-/*
- * 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;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.File;
-import java.lang.reflect.Method;
-import java.text.SimpleDateFormat;
-import java.util.List;
-import java.util.TimeZone;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.AbstractKylinTestCase;
-import org.apache.kylin.common.util.ClassUtil;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.CubeUpdate;
-import org.apache.kylin.engine.EngineFactory;
-import org.apache.kylin.engine.mr.CubingJob;
-import org.apache.kylin.job.engine.JobEngineConfig;
-import org.apache.kylin.job.execution.AbstractExecutable;
-import org.apache.kylin.job.execution.DefaultChainedExecutable;
-import org.apache.kylin.job.execution.ExecutableState;
-import org.apache.kylin.job.impl.threadpool.DefaultScheduler;
-import org.apache.kylin.job.manager.ExecutableManager;
-import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
-import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
-import org.apache.kylin.storage.hbase.util.ZookeeperJobLock;
-
-import com.google.common.collect.Lists;
-
-public class BuildCubeWithEngine {
-
- private CubeManager cubeManager;
- private DefaultScheduler scheduler;
- protected ExecutableManager jobService;
- private static boolean fastBuildMode = false;
-
- private static final Log logger = LogFactory.getLog(BuildCubeWithEngine.class);
-
- public static void main(String[] args) throws Exception {
- beforeClass();
- BuildCubeWithEngine buildCubeWithEngine = new BuildCubeWithEngine();
- buildCubeWithEngine.before();
- buildCubeWithEngine.build();
- afterClass();
- }
-
- public static void beforeClass() throws Exception {
- logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
- ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
-
- String fastModeStr = System.getProperty("fastBuildMode");
- if (fastModeStr != null && fastModeStr.equalsIgnoreCase("true")) {
- fastBuildMode = true;
- logger.info("Will use fast build mode");
- } else {
- logger.info("Will not use fast build mode");
- }
-
- System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
- if (System.getProperty("hdp.version") == null) {
- throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
- }
-
- HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
- }
-
- public void before() throws Exception {
-
- DeployUtil.initCliWorkDir();
- DeployUtil.deployMetadata();
- DeployUtil.overrideJobJarLocations();
-
- final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
- jobService = ExecutableManager.getInstance(kylinConfig);
- scheduler = DefaultScheduler.getInstance();
- scheduler.init(new JobEngineConfig(kylinConfig), new ZookeeperJobLock());
- if (!scheduler.hasStarted()) {
- throw new RuntimeException("scheduler has not been started");
- }
- cubeManager = CubeManager.getInstance(kylinConfig);
- for (String jobId : jobService.getAllJobIds()) {
- if (jobService.getJob(jobId) instanceof CubingJob) {
- jobService.deleteJob(jobId);
- }
- }
-
- }
-
- public static void afterClass() {
- HBaseMetadataTestCase.staticCleanupTestMetadata();
- }
-
- public void build() throws Exception {
- DeployUtil.prepareTestDataForNormalCubes("test_kylin_cube_with_slr_left_join_empty");
- testInner();
- testLeft();
- }
-
- protected void waitForJob(String jobId) {
- while (true) {
- AbstractExecutable job = jobService.getJob(jobId);
- if (job.getStatus() == ExecutableState.SUCCEED || job.getStatus() == ExecutableState.ERROR) {
- break;
- } else {
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- }
- }
-
- private void testInner() throws Exception {
- String[] testCase = new String[] { "testInnerJoinCubeWithoutSlr", "testInnerJoinCubeWithSlr" };
- runTestAndAssertSucceed(testCase);
- }
-
- private void testLeft() throws Exception {
- String[] testCase = new String[] { "testLeftJoinCubeWithSlr", "testLeftJoinCubeWithoutSlr" };
- runTestAndAssertSucceed(testCase);
- }
-
- private void runTestAndAssertSucceed(String[] testCase) throws Exception {
- ExecutorService executorService = Executors.newFixedThreadPool(testCase.length);
- final CountDownLatch countDownLatch = new CountDownLatch(testCase.length);
- List<Future<List<String>>> tasks = Lists.newArrayListWithExpectedSize(testCase.length);
- for (int i = 0; i < testCase.length; i++) {
- tasks.add(executorService.submit(new TestCallable(testCase[i], countDownLatch)));
- }
- countDownLatch.await();
- try {
- for (int i = 0; i < tasks.size(); ++i) {
- Future<List<String>> task = tasks.get(i);
- final List<String> jobIds = task.get();
- for (String jobId : jobIds) {
- assertJobSucceed(jobId);
- }
- }
- } catch (Exception ex) {
- logger.error(ex);
- throw ex;
- }
- }
-
- private void assertJobSucceed(String jobId) {
- assertEquals("The job '" + jobId + "' is failed.", ExecutableState.SUCCEED, jobService.getOutput(jobId).getState());
- }
-
- private class TestCallable implements Callable<List<String>> {
-
- private final String methodName;
- private final CountDownLatch countDownLatch;
-
- public TestCallable(String methodName, CountDownLatch countDownLatch) {
- this.methodName = methodName;
- this.countDownLatch = countDownLatch;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public List<String> call() throws Exception {
- try {
- final Method method = BuildCubeWithEngine.class.getDeclaredMethod(methodName);
- method.setAccessible(true);
- return (List<String>) method.invoke(BuildCubeWithEngine.this);
- } catch (Exception e) {
- logger.error(e.getMessage());
- throw e;
- } finally {
- countDownLatch.countDown();
- }
- }
- }
-
- @SuppressWarnings("unused")
- // called by reflection
- private List<String> testInnerJoinCubeWithSlr() throws Exception {
- clearSegment("test_kylin_cube_with_slr_empty");
- SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
- f.setTimeZone(TimeZone.getTimeZone("GMT"));
- long date1 = 0;
- long date2 = f.parse("2015-01-01").getTime();
- long date3 = f.parse("2022-01-01").getTime();
- List<String> result = Lists.newArrayList();
-
- if (fastBuildMode) {
- result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date3));
- } else {
- result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date2));
- result.add(buildSegment("test_kylin_cube_with_slr_empty", date2, date3));//empty segment
- }
- return result;
- }
-
- @SuppressWarnings("unused")
- // called by reflection
- private List<String> testInnerJoinCubeWithoutSlr() throws Exception {
-
- clearSegment("test_kylin_cube_without_slr_empty");
- SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
- f.setTimeZone(TimeZone.getTimeZone("GMT"));
- long date1 = 0;
- long date2 = f.parse("2013-01-01").getTime();
- long date3 = f.parse("2013-07-01").getTime();
- long date4 = f.parse("2022-01-01").getTime();
- List<String> result = Lists.newArrayList();
-
- if (fastBuildMode) {
- result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date4));
- } else {
- result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date2));
- result.add(buildSegment("test_kylin_cube_without_slr_empty", date2, date3));
- result.add(buildSegment("test_kylin_cube_without_slr_empty", date3, date4));
- result.add(mergeSegment("test_kylin_cube_without_slr_empty", date1, date3));//don't merge all segments
- }
- return result;
-
- }
-
- @SuppressWarnings("unused")
- // called by reflection
- private List<String> testLeftJoinCubeWithoutSlr() throws Exception {
- SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
- f.setTimeZone(TimeZone.getTimeZone("GMT"));
- List<String> result = Lists.newArrayList();
- final String cubeName = "test_kylin_cube_without_slr_left_join_empty";
- clearSegment(cubeName);
-
- long date1 = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
- long date2 = f.parse("2012-06-01").getTime();
- long date3 = f.parse("2022-01-01").getTime();
- long date4 = f.parse("2023-01-01").getTime();
-
- if (fastBuildMode) {
- result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date4));
- } else {
- result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date2));
- result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date2, date3));
- result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date3, date4));//empty segment
- result.add(mergeSegment("test_kylin_cube_without_slr_left_join_empty", date1, date3));//don't merge all segments
- }
-
- return result;
-
- }
-
- @SuppressWarnings("unused")
- // called by reflection
- private List<String> testLeftJoinCubeWithSlr() throws Exception {
- String cubeName = "test_kylin_cube_with_slr_left_join_empty";
- clearSegment(cubeName);
-
- SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
- f.setTimeZone(TimeZone.getTimeZone("GMT"));
- long date1 = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
- long date2 = f.parse("2013-01-01").getTime();
- long date3 = f.parse("2013-07-01").getTime();
- long date4 = f.parse("2022-01-01").getTime();
-
- List<String> result = Lists.newArrayList();
- if (fastBuildMode) {
- result.add(buildSegment(cubeName, date1, date4));
- } else {
- result.add(buildSegment(cubeName, date1, date2));
- result.add(buildSegment(cubeName, date2, date3));
- result.add(buildSegment(cubeName, date3, date4));
- result.add(mergeSegment(cubeName, date1, date3));//don't merge all segments
- }
- return result;
-
- }
-
- private void clearSegment(String cubeName) throws Exception {
- CubeInstance cube = cubeManager.getCube(cubeName);
- // remove all existing segments
- CubeUpdate cubeBuilder = new CubeUpdate(cube);
- cubeBuilder.setToRemoveSegs(cube.getSegments().toArray(new CubeSegment[cube.getSegments().size()]));
- cubeManager.updateCube(cubeBuilder);
- }
-
- private String mergeSegment(String cubeName, long startDate, long endDate) throws Exception {
- CubeSegment segment = cubeManager.mergeSegments(cubeManager.getCube(cubeName), startDate, endDate, true);
- DefaultChainedExecutable job = EngineFactory.createBatchMergeJob(segment, "TEST");
- jobService.addJob(job);
- waitForJob(job.getId());
- return job.getId();
- }
-
- private String buildSegment(String cubeName, long startDate, long endDate) throws Exception {
- CubeSegment segment = cubeManager.appendSegments(cubeManager.getCube(cubeName), endDate);
- DefaultChainedExecutable job = EngineFactory.createBatchCubingJob(segment, "TEST");
- jobService.addJob(job);
- waitForJob(job.getId());
- return job.getId();
- }
-
- private int cleanupOldStorage() throws Exception {
- String[] args = { "--delete", "true" };
-
- int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
- return exitCode;
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
new file mode 100644
index 0000000..331e21d
--- /dev/null
+++ b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithEngineTest.java
@@ -0,0 +1,333 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.CubeUpdate;
+import org.apache.kylin.engine.EngineFactory;
+import org.apache.kylin.engine.mr.CubingJob;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.impl.threadpool.DefaultScheduler;
+import org.apache.kylin.job.manager.ExecutableManager;
+import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
+import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
+import org.apache.kylin.storage.hbase.util.ZookeeperJobLock;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+public class BuildCubeWithEngineTest {
+
+ private CubeManager cubeManager;
+ private DefaultScheduler scheduler;
+ protected ExecutableManager jobService;
+ private static boolean fastBuildMode = false;
+
+ private static final Log logger = LogFactory.getLog(BuildCubeWithEngineTest.class);
+
+ protected void waitForJob(String jobId) {
+ while (true) {
+ AbstractExecutable job = jobService.getJob(jobId);
+ if (job.getStatus() == ExecutableState.SUCCEED || job.getStatus() == ExecutableState.ERROR) {
+ break;
+ } else {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+
+ String fastModeStr = System.getProperty("fastBuildMode");
+ if (fastModeStr != null && fastModeStr.equalsIgnoreCase("true")) {
+ fastBuildMode = true;
+ logger.info("Will use fast build mode");
+ } else {
+ logger.info("Will not use fast build mode");
+ }
+
+ System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+
+ }
+
+ @Before
+ public void before() throws Exception {
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+
+ DeployUtil.initCliWorkDir();
+ DeployUtil.deployMetadata();
+ DeployUtil.overrideJobJarLocations();
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ jobService = ExecutableManager.getInstance(kylinConfig);
+ scheduler = DefaultScheduler.getInstance();
+ scheduler.init(new JobEngineConfig(kylinConfig), new ZookeeperJobLock());
+ if (!scheduler.hasStarted()) {
+ throw new RuntimeException("scheduler has not been started");
+ }
+ cubeManager = CubeManager.getInstance(kylinConfig);
+ for (String jobId : jobService.getAllJobIds()) {
+ if (jobService.getJob(jobId) instanceof CubingJob) {
+ jobService.deleteJob(jobId);
+ }
+ }
+
+ }
+
+ @After
+ public void after() {
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
+ }
+
+ @Test
+ public void test() throws Exception {
+ DeployUtil.prepareTestDataForNormalCubes("test_kylin_cube_with_slr_left_join_empty");
+ testInner();
+ testLeft();
+ }
+
+ private void testInner() throws Exception {
+ String[] testCase = new String[] { "testInnerJoinCubeWithoutSlr", "testInnerJoinCubeWithSlr" };
+ runTestAndAssertSucceed(testCase);
+ }
+
+ private void testLeft() throws Exception {
+ String[] testCase = new String[] { "testLeftJoinCubeWithSlr", "testLeftJoinCubeWithoutSlr" };
+ runTestAndAssertSucceed(testCase);
+ }
+
+ private void runTestAndAssertSucceed(String[] testCase) throws Exception {
+ ExecutorService executorService = Executors.newFixedThreadPool(testCase.length);
+ final CountDownLatch countDownLatch = new CountDownLatch(testCase.length);
+ List<Future<List<String>>> tasks = Lists.newArrayListWithExpectedSize(testCase.length);
+ for (int i = 0; i < testCase.length; i++) {
+ tasks.add(executorService.submit(new TestCallable(testCase[i], countDownLatch)));
+ }
+ countDownLatch.await();
+ try {
+ for (int i = 0; i < tasks.size(); ++i) {
+ Future<List<String>> task = tasks.get(i);
+ final List<String> jobIds = task.get();
+ for (String jobId : jobIds) {
+ assertJobSucceed(jobId);
+ }
+ }
+ } catch (Exception ex) {
+ logger.error(ex);
+ throw ex;
+ }
+ }
+
+ private void assertJobSucceed(String jobId) {
+ assertEquals("The job '" + jobId + "' is failed.", ExecutableState.SUCCEED, jobService.getOutput(jobId).getState());
+ }
+
+ private class TestCallable implements Callable<List<String>> {
+
+ private final String methodName;
+ private final CountDownLatch countDownLatch;
+
+ public TestCallable(String methodName, CountDownLatch countDownLatch) {
+ this.methodName = methodName;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<String> call() throws Exception {
+ try {
+ final Method method = BuildCubeWithEngineTest.class.getDeclaredMethod(methodName);
+ method.setAccessible(true);
+ return (List<String>) method.invoke(BuildCubeWithEngineTest.this);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ throw e;
+ } finally {
+ countDownLatch.countDown();
+ }
+ }
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testInnerJoinCubeWithSlr() throws Exception {
+ clearSegment("test_kylin_cube_with_slr_empty");
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long date1 = 0;
+ long date2 = f.parse("2015-01-01").getTime();
+ long date3 = f.parse("2022-01-01").getTime();
+ List<String> result = Lists.newArrayList();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date3));
+ } else {
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date2, date3));//empty segment
+ }
+ return result;
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testInnerJoinCubeWithoutSlr() throws Exception {
+
+ clearSegment("test_kylin_cube_without_slr_empty");
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long date1 = 0;
+ long date2 = f.parse("2013-01-01").getTime();
+ long date3 = f.parse("2013-07-01").getTime();
+ long date4 = f.parse("2022-01-01").getTime();
+ List<String> result = Lists.newArrayList();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date4));
+ } else {
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date2, date3));
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date3, date4));
+ result.add(mergeSegment("test_kylin_cube_without_slr_empty", date1, date3));//don't merge all segments
+ }
+ return result;
+
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testLeftJoinCubeWithoutSlr() throws Exception {
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ List<String> result = Lists.newArrayList();
+ final String cubeName = "test_kylin_cube_without_slr_left_join_empty";
+ clearSegment(cubeName);
+
+ long date1 = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
+ long date2 = f.parse("2012-06-01").getTime();
+ long date3 = f.parse("2022-01-01").getTime();
+ long date4 = f.parse("2023-01-01").getTime();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date4));
+ } else {
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date2, date3));
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date3, date4));//empty segment
+ result.add(mergeSegment("test_kylin_cube_without_slr_left_join_empty", date1, date3));//don't merge all segments
+ }
+
+ return result;
+
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testLeftJoinCubeWithSlr() throws Exception {
+ String cubeName = "test_kylin_cube_with_slr_left_join_empty";
+ clearSegment(cubeName);
+
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long date1 = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
+ long date2 = f.parse("2013-01-01").getTime();
+ long date3 = f.parse("2013-07-01").getTime();
+ long date4 = f.parse("2022-01-01").getTime();
+
+ List<String> result = Lists.newArrayList();
+ if (fastBuildMode) {
+ result.add(buildSegment(cubeName, date1, date4));
+ } else {
+ result.add(buildSegment(cubeName, date1, date2));
+ result.add(buildSegment(cubeName, date2, date3));
+ result.add(buildSegment(cubeName, date3, date4));
+ result.add(mergeSegment(cubeName, date1, date3));//don't merge all segments
+ }
+ return result;
+
+ }
+
+ private void clearSegment(String cubeName) throws Exception {
+ CubeInstance cube = cubeManager.getCube(cubeName);
+ // remove all existing segments
+ CubeUpdate cubeBuilder = new CubeUpdate(cube);
+ cubeBuilder.setToRemoveSegs(cube.getSegments().toArray(new CubeSegment[cube.getSegments().size()]));
+ cubeManager.updateCube(cubeBuilder);
+ }
+
+ private String mergeSegment(String cubeName, long startDate, long endDate) throws Exception {
+ CubeSegment segment = cubeManager.mergeSegments(cubeManager.getCube(cubeName), startDate, endDate, true);
+ DefaultChainedExecutable job = EngineFactory.createBatchMergeJob(segment, "TEST");
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
+ private String buildSegment(String cubeName, long startDate, long endDate) throws Exception {
+ CubeSegment segment = cubeManager.appendSegments(cubeManager.getCube(cubeName), endDate);
+ DefaultChainedExecutable job = EngineFactory.createBatchCubingJob(segment, "TEST");
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
+ private int cleanupOldStorage() throws Exception {
+ String[] args = { "--delete", "true" };
+
+ int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
+ return exitCode;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStream.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStream.java b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStream.java
deleted file mode 100644
index d1f0a48..0000000
--- a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStream.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- *
- *
- * 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;
-
-import java.io.File;
-import java.util.UUID;
-
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.AbstractKylinTestCase;
-import org.apache.kylin.common.util.ClassUtil;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.engine.streaming.OneOffStreamingBuilder;
-import org.apache.kylin.engine.streaming.StreamingConfig;
-import org.apache.kylin.engine.streaming.StreamingManager;
-import org.apache.kylin.source.kafka.KafkaConfigManager;
-import org.apache.kylin.source.kafka.config.KafkaConfig;
-import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * for streaming cubing case "test_streaming_table"
- */
-public class BuildCubeWithStream {
-
- private static final Logger logger = LoggerFactory.getLogger(BuildCubeWithStream.class);
- private static final String streamingName = "test_streaming_table_cube";
- private static final long startTime = DateFormat.stringToMillis("2015-01-01 00:00:00");
- private static final long endTime = DateFormat.stringToMillis("2015-01-03 00:00:00");
- private static final long batchInterval = 16 * 60 * 60 * 1000;//16 hours
-
- private KylinConfig kylinConfig;
-
- public static void main(String[] args) throws Exception {
- beforeClass();
- BuildCubeWithStream buildCubeWithStream = new BuildCubeWithStream();
- buildCubeWithStream.before();
- buildCubeWithStream.build();
- afterClass();
- }
-
- public static void beforeClass() throws Exception {
- logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
- ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
- System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
- if (System.getProperty("hdp.version") == null) {
- throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
- }
- HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
- }
-
- public void before() throws Exception {
- DeployUtil.overrideJobJarLocations();
-
- kylinConfig = KylinConfig.getInstanceFromEnv();
-
- final StreamingConfig config = StreamingManager.getInstance(KylinConfig.getInstanceFromEnv()).getStreamingConfig(streamingName);
-
- //Use a random topic for kafka data stream
- KafkaConfig streamingConfig = KafkaConfigManager.getInstance(kylinConfig).getKafkaConfig(streamingName);
- streamingConfig.setTopic(UUID.randomUUID().toString());
- KafkaConfigManager.getInstance(kylinConfig).saveKafkaConfig(streamingConfig);
-
- DeployUtil.prepareTestDataForStreamingCube(startTime, endTime, config.getCubeName(), streamingConfig);
- }
-
- public static void afterClass() throws Exception {
- HBaseMetadataTestCase.staticCleanupTestMetadata();
- }
-
- public void build() throws Exception {
- logger.info("start time:" + startTime + " end time:" + endTime + " batch interval:" + batchInterval + " batch count:" + ((endTime - startTime) / batchInterval));
- for (long start = startTime; start < endTime; start += batchInterval) {
- logger.info(String.format("build batch:{%d, %d}", start, start + batchInterval));
- new OneOffStreamingBuilder(streamingName, start, start + batchInterval).build().run();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStreamTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStreamTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStreamTest.java
new file mode 100644
index 0000000..9e60622
--- /dev/null
+++ b/assembly/src/test/java/org/apache/kylin/job/BuildCubeWithStreamTest.java
@@ -0,0 +1,114 @@
+/*
+ *
+ *
+ * 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;
+
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.engine.streaming.OneOffStreamingBuilder;
+import org.apache.kylin.engine.streaming.StreamingConfig;
+import org.apache.kylin.engine.streaming.StreamingManager;
+import org.apache.kylin.source.kafka.KafkaConfigManager;
+import org.apache.kylin.source.kafka.config.KafkaConfig;
+import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
+import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+
+/**
+ * for streaming cubing case "test_streaming_table"
+ */
+public class BuildCubeWithStreamTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(BuildCubeWithStreamTest.class);
+ private static final String streamingName = "test_streaming_table_cube";
+ private static final long startTime = DateFormat.stringToMillis("2015-01-01 00:00:00");
+ private static final long endTime = DateFormat.stringToMillis("2015-01-03 00:00:00");
+ private static final long batchInterval = 16 * 60 * 60 * 1000;//16 hours
+
+ private KylinConfig kylinConfig;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+
+ }
+
+ @Before
+ public void before() throws Exception {
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+ DeployUtil.overrideJobJarLocations();
+
+ kylinConfig = KylinConfig.getInstanceFromEnv();
+
+ final StreamingConfig config = StreamingManager.getInstance(KylinConfig.getInstanceFromEnv()).getStreamingConfig(streamingName);
+
+ //Use a random topic for kafka data stream
+ KafkaConfig streamingConfig = KafkaConfigManager.getInstance(kylinConfig).getKafkaConfig(streamingName);
+ streamingConfig.setTopic(UUID.randomUUID().toString());
+ KafkaConfigManager.getInstance(kylinConfig).saveKafkaConfig(streamingConfig);
+
+ DeployUtil.prepareTestDataForStreamingCube(startTime, endTime, config.getCubeName(), streamingConfig);
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
+ }
+
+ @Test
+ public void test() throws Exception {
+ logger.info("start time:" + startTime + " end time:" + endTime + " batch interval:" + batchInterval + " batch count:" + ((endTime - startTime) / batchInterval));
+ for (long start = startTime; start < endTime; start += batchInterval) {
+ logger.info(String.format("build batch:{%d, %d}", start, start + batchInterval));
+ new OneOffStreamingBuilder(streamingName, start, start + batchInterval).build().run();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildIIWithEngineTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithEngineTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithEngineTest.java
index 31dc396..0158fad 100644
--- a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithEngineTest.java
+++ b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithEngineTest.java
@@ -38,7 +38,6 @@ import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.AbstractKylinTestCase;
import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.engine.mr.invertedindex.BatchIIJobBuilder;
-import org.apache.kylin.engine.mr.invertedindex.IIJob;
import org.apache.kylin.invertedindex.IIInstance;
import org.apache.kylin.invertedindex.IIManager;
import org.apache.kylin.invertedindex.IISegment;
@@ -46,6 +45,7 @@ import org.apache.kylin.job.engine.JobEngineConfig;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableState;
import org.apache.kylin.job.impl.threadpool.DefaultScheduler;
+import org.apache.kylin.engine.mr.invertedindex.IIJob;
import org.apache.kylin.job.manager.ExecutableManager;
import org.apache.kylin.metadata.realization.RealizationStatusEnum;
import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
@@ -59,8 +59,9 @@ import org.junit.Test;
import com.google.common.collect.Lists;
-//TODO: convert it to a normal class rather than a test case, like in BuildCubeWithEngine
-@Ignore
+/**
+ * @author shaoshi
+ */
public class BuildIIWithEngineTest {
private JobEngineConfig jobEngineConfig;
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStream.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStream.java b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStream.java
deleted file mode 100644
index 1e38d3a..0000000
--- a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStream.java
+++ /dev/null
@@ -1,292 +0,0 @@
-/*
- *
- *
- * 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;
-
-import java.io.File;
-import java.io.IOException;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-import java.util.TimeZone;
-import java.util.UUID;
-
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.AbstractKylinTestCase;
-import org.apache.kylin.common.util.ClassUtil;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.common.util.StreamingBatch;
-import org.apache.kylin.common.util.StreamingMessage;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
-import org.apache.kylin.invertedindex.IIInstance;
-import org.apache.kylin.invertedindex.IIManager;
-import org.apache.kylin.invertedindex.IISegment;
-import org.apache.kylin.invertedindex.index.Slice;
-import org.apache.kylin.invertedindex.index.SliceBuilder;
-import org.apache.kylin.invertedindex.model.IIDesc;
-import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc;
-import org.apache.kylin.invertedindex.model.IIKeyValueCodec;
-import org.apache.kylin.invertedindex.model.IIRow;
-import org.apache.kylin.job.common.ShellExecutable;
-import org.apache.kylin.job.constant.ExecutableConstants;
-import org.apache.kylin.job.engine.JobEngineConfig;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.realization.RealizationStatusEnum;
-import org.apache.kylin.source.hive.HiveTableReader;
-import org.apache.kylin.storage.hbase.HBaseConnection;
-import org.apache.kylin.storage.hbase.ii.IICreateHTableJob;
-import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
-import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
-
-public class BuildIIWithStream {
-
- private static final Logger logger = LoggerFactory.getLogger(BuildIIWithStream.class);
-
- private static final String[] II_NAME = new String[] { "test_kylin_ii_left_join", "test_kylin_ii_inner_join" };
- private IIManager iiManager;
- private KylinConfig kylinConfig;
-
- public static void main(String[] args) throws Exception {
- beforeClass();
- BuildIIWithStream buildCubeWithEngine = new BuildIIWithStream();
- buildCubeWithEngine.before();
- buildCubeWithEngine.build();
- afterClass();
- }
-
- public static void beforeClass() throws Exception {
- logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
- ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
- if (System.getProperty("hdp.version") == null) {
- throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
- }
- HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
- }
-
- public void before() throws Exception {
- DeployUtil.overrideJobJarLocations();
-
- kylinConfig = KylinConfig.getInstanceFromEnv();
- iiManager = IIManager.getInstance(kylinConfig);
- for (String iiInstance : II_NAME) {
-
- IIInstance ii = iiManager.getII(iiInstance);
- if (ii.getStatus() != RealizationStatusEnum.DISABLED) {
- ii.setStatus(RealizationStatusEnum.DISABLED);
- iiManager.updateII(ii);
- }
- }
- }
-
- public static void afterClass() throws Exception {
- cleanupOldStorage();
- HBaseMetadataTestCase.staticCleanupTestMetadata();
- }
-
- private String createIntermediateTable(IIDesc desc, KylinConfig kylinConfig) throws IOException {
- IIJoinedFlatTableDesc intermediateTableDesc = new IIJoinedFlatTableDesc(desc);
- JobEngineConfig jobEngineConfig = new JobEngineConfig(kylinConfig);
- final String uuid = UUID.randomUUID().toString();
- final String useDatabaseHql = "USE " + kylinConfig.getHiveDatabaseForIntermediateTable() + ";";
- final String dropTableHql = JoinedFlatTable.generateDropTableStatement(intermediateTableDesc);
- final String createTableHql = JoinedFlatTable.generateCreateTableStatement(intermediateTableDesc, JobBuilderSupport.getJobWorkingDir(jobEngineConfig, uuid));
- String insertDataHqls;
- try {
- insertDataHqls = JoinedFlatTable.generateInsertDataStatement(intermediateTableDesc, jobEngineConfig);
- } catch (IOException e1) {
- e1.printStackTrace();
- throw new RuntimeException("Failed to generate insert data SQL for intermediate table.");
- }
-
- ShellExecutable step = new ShellExecutable();
- StringBuffer buf = new StringBuffer();
- buf.append("hive -e \"");
- buf.append(useDatabaseHql + "\n");
- buf.append(dropTableHql + "\n");
- buf.append(createTableHql + "\n");
- buf.append(insertDataHqls + "\n");
- buf.append("\"");
-
- step.setCmd(buf.toString());
- logger.info(step.getCmd());
- step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
- kylinConfig.getCliCommandExecutor().execute(step.getCmd(), null);
- return intermediateTableDesc.getTableName();
- }
-
- private void clearSegment(String iiName) throws Exception {
- IIInstance ii = iiManager.getII(iiName);
- ii.getSegments().clear();
- iiManager.updateII(ii);
- }
-
- private IISegment createSegment(String iiName) throws Exception {
- clearSegment(iiName);
- SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
- f.setTimeZone(TimeZone.getTimeZone("GMT"));
-
- long date1 = 0;
- long date2 = f.parse("2015-01-01").getTime();
- return buildSegment(iiName, date1, date2);
- }
-
- private IISegment buildSegment(String iiName, long startDate, long endDate) throws Exception {
- IIInstance iiInstance = iiManager.getII(iiName);
- IISegment segment = iiManager.buildSegment(iiInstance, startDate, endDate);
- iiInstance.getSegments().add(segment);
- iiManager.updateII(iiInstance);
- return segment;
- }
-
- private void buildII(String iiName) throws Exception {
- final IIDesc desc = iiManager.getII(iiName).getDescriptor();
- final String tableName = createIntermediateTable(desc, kylinConfig);
- logger.info("intermediate table name:" + tableName);
-
- HiveTableReader reader = new HiveTableReader("default", tableName);
- final List<TblColRef> tblColRefs = desc.listAllColumns();
- for (TblColRef tblColRef : tblColRefs) {
- if (desc.isMetricsCol(tblColRef)) {
- logger.info("matrix:" + tblColRef.getName());
- } else {
- logger.info("measure:" + tblColRef.getName());
- }
- }
- final IISegment segment = createSegment(iiName);
- final HTableInterface htable = HBaseConnection.get(KylinConfig.getInstanceFromEnv().getStorageUrl()).getTable(segment.getStorageLocationIdentifier());
- String[] args = new String[] { "-iiname", iiName, "-htablename", segment.getStorageLocationIdentifier() };
- ToolRunner.run(new IICreateHTableJob(), args);
-
- final IIDesc iiDesc = segment.getIIDesc();
- final SliceBuilder sliceBuilder = new SliceBuilder(desc, (short) 0);
-
- List<String[]> sorted = getSortedRows(reader, desc.getTimestampColumn());
- int count = sorted.size();
- ArrayList<StreamingMessage> messages = Lists.newArrayList();
- for (String[] row : sorted) {
- messages.add((parse(row)));
- if (messages.size() >= iiDesc.getSliceSize()) {
- build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable);
- messages.clear();
- }
- }
-
- if (!messages.isEmpty()) {
- build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable);
- }
-
- reader.close();
- logger.info("total record count:" + count + " htable:" + segment.getStorageLocationIdentifier());
- logger.info("stream build finished, htable name:" + segment.getStorageLocationIdentifier());
- }
-
- public void build() throws Exception {
- for (String iiName : II_NAME) {
- buildII(iiName);
- IIInstance ii = iiManager.getII(iiName);
- if (ii.getStatus() != RealizationStatusEnum.READY) {
- ii.setStatus(RealizationStatusEnum.READY);
- iiManager.updateII(ii);
- }
- }
- }
-
- private void build(SliceBuilder sliceBuilder, StreamingBatch batch, HTableInterface htable) throws IOException {
- final Slice slice = sliceBuilder.buildSlice(batch);
- try {
- loadToHBase(htable, slice, new IIKeyValueCodec(slice.getInfo()));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- private void loadToHBase(HTableInterface hTable, Slice slice, IIKeyValueCodec codec) throws IOException {
- List<Put> data = Lists.newArrayList();
- for (IIRow row : codec.encodeKeyValue(slice)) {
- final byte[] key = row.getKey().get();
- final byte[] value = row.getValue().get();
- Put put = new Put(key);
- put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_QUALIFIER_BYTES, value);
- final ImmutableBytesWritable dictionary = row.getDictionary();
- final byte[] dictBytes = dictionary.get();
- if (dictionary.getOffset() == 0 && dictionary.getLength() == dictBytes.length) {
- put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_DICTIONARY_BYTES, dictBytes);
- } else {
- throw new RuntimeException("dict offset should be 0, and dict length should be " + dictBytes.length + " but they are" + dictionary.getOffset() + " " + dictionary.getLength());
- }
- data.add(put);
- }
- hTable.put(data);
- //omit hTable.flushCommits(), because htable is auto flush
- }
-
- private StreamingMessage parse(String[] row) {
- return new StreamingMessage(Lists.newArrayList(row), System.currentTimeMillis(), System.currentTimeMillis(), Collections.<String, Object> emptyMap());
- }
-
- private List<String[]> getSortedRows(HiveTableReader reader, final int tsCol) throws IOException {
- List<String[]> unsorted = Lists.newArrayList();
- while (reader.next()) {
- unsorted.add(reader.getRow());
- }
- Collections.sort(unsorted, new Comparator<String[]>() {
- @Override
- public int compare(String[] o1, String[] o2) {
- long t1 = DateFormat.stringToMillis(o1[tsCol]);
- long t2 = DateFormat.stringToMillis(o2[tsCol]);
- return Long.compare(t1, t2);
- }
- });
- return unsorted;
- }
-
- private static int cleanupOldStorage() throws Exception {
- String[] args = { "--delete", "true" };
-
- int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
- return exitCode;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
new file mode 100644
index 0000000..2317e62
--- /dev/null
+++ b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
@@ -0,0 +1,290 @@
+/*
+ *
+ *
+ * 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;
+
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.engine.mr.JobBuilderSupport;
+import org.apache.kylin.common.util.StreamingBatch;
+import org.apache.kylin.common.util.StreamingMessage;
+import org.apache.kylin.invertedindex.IIInstance;
+import org.apache.kylin.invertedindex.IIManager;
+import org.apache.kylin.invertedindex.IISegment;
+import org.apache.kylin.invertedindex.index.Slice;
+import org.apache.kylin.invertedindex.model.IIDesc;
+import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc;
+import org.apache.kylin.invertedindex.model.IIKeyValueCodec;
+import org.apache.kylin.invertedindex.model.IIRow;
+import org.apache.kylin.invertedindex.index.SliceBuilder;
+import org.apache.kylin.job.common.ShellExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.storage.hbase.ii.IICreateHTableJob;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.realization.RealizationStatusEnum;
+import org.apache.kylin.source.hive.HiveTableReader;
+import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
+import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+/**
+ */
+public class BuildIIWithStreamTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(BuildIIWithStreamTest.class);
+
+ private static final String[] II_NAME = new String[] { "test_kylin_ii_left_join", "test_kylin_ii_inner_join" };
+ private IIManager iiManager;
+ private KylinConfig kylinConfig;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+ }
+
+ @Before
+ public void before() throws Exception {
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+ DeployUtil.overrideJobJarLocations();
+
+ kylinConfig = KylinConfig.getInstanceFromEnv();
+ iiManager = IIManager.getInstance(kylinConfig);
+ for (String iiInstance : II_NAME) {
+
+ IIInstance ii = iiManager.getII(iiInstance);
+ if (ii.getStatus() != RealizationStatusEnum.DISABLED) {
+ ii.setStatus(RealizationStatusEnum.DISABLED);
+ iiManager.updateII(ii);
+ }
+ }
+ }
+
+ @AfterClass
+ public static void cleanup() throws Exception {
+ cleanupOldStorage();
+ }
+
+ private String createIntermediateTable(IIDesc desc, KylinConfig kylinConfig) throws IOException {
+ IIJoinedFlatTableDesc intermediateTableDesc = new IIJoinedFlatTableDesc(desc);
+ JobEngineConfig jobEngineConfig = new JobEngineConfig(kylinConfig);
+ final String uuid = UUID.randomUUID().toString();
+ final String useDatabaseHql = "USE " + kylinConfig.getHiveDatabaseForIntermediateTable() + ";";
+ final String dropTableHql = JoinedFlatTable.generateDropTableStatement(intermediateTableDesc);
+ final String createTableHql = JoinedFlatTable.generateCreateTableStatement(intermediateTableDesc, JobBuilderSupport.getJobWorkingDir(jobEngineConfig, uuid));
+ String insertDataHqls;
+ try {
+ insertDataHqls = JoinedFlatTable.generateInsertDataStatement(intermediateTableDesc, jobEngineConfig);
+ } catch (IOException e1) {
+ e1.printStackTrace();
+ throw new RuntimeException("Failed to generate insert data SQL for intermediate table.");
+ }
+
+ ShellExecutable step = new ShellExecutable();
+ StringBuffer buf = new StringBuffer();
+ buf.append("hive -e \"");
+ buf.append(useDatabaseHql + "\n");
+ buf.append(dropTableHql + "\n");
+ buf.append(createTableHql + "\n");
+ buf.append(insertDataHqls + "\n");
+ buf.append("\"");
+
+ step.setCmd(buf.toString());
+ logger.info(step.getCmd());
+ step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
+ kylinConfig.getCliCommandExecutor().execute(step.getCmd(), null);
+ return intermediateTableDesc.getTableName();
+ }
+
+ private void clearSegment(String iiName) throws Exception {
+ IIInstance ii = iiManager.getII(iiName);
+ ii.getSegments().clear();
+ iiManager.updateII(ii);
+ }
+
+ private IISegment createSegment(String iiName) throws Exception {
+ clearSegment(iiName);
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+
+ long date1 = 0;
+ long date2 = f.parse("2015-01-01").getTime();
+ return buildSegment(iiName, date1, date2);
+ }
+
+ private IISegment buildSegment(String iiName, long startDate, long endDate) throws Exception {
+ IIInstance iiInstance = iiManager.getII(iiName);
+ IISegment segment = iiManager.buildSegment(iiInstance, startDate, endDate);
+ iiInstance.getSegments().add(segment);
+ iiManager.updateII(iiInstance);
+ return segment;
+ }
+
+ private void buildII(String iiName) throws Exception {
+ final IIDesc desc = iiManager.getII(iiName).getDescriptor();
+ final String tableName = createIntermediateTable(desc, kylinConfig);
+ logger.info("intermediate table name:" + tableName);
+
+ HiveTableReader reader = new HiveTableReader("default", tableName);
+ final List<TblColRef> tblColRefs = desc.listAllColumns();
+ for (TblColRef tblColRef : tblColRefs) {
+ if (desc.isMetricsCol(tblColRef)) {
+ logger.info("matrix:" + tblColRef.getName());
+ } else {
+ logger.info("measure:" + tblColRef.getName());
+ }
+ }
+ final IISegment segment = createSegment(iiName);
+ final HTableInterface htable = HBaseConnection.get(KylinConfig.getInstanceFromEnv().getStorageUrl()).getTable(segment.getStorageLocationIdentifier());
+ String[] args = new String[] { "-iiname", iiName, "-htablename", segment.getStorageLocationIdentifier() };
+ ToolRunner.run(new IICreateHTableJob(), args);
+
+ final IIDesc iiDesc = segment.getIIDesc();
+ final SliceBuilder sliceBuilder = new SliceBuilder(desc, (short) 0);
+
+ List<String[]> sorted = getSortedRows(reader, desc.getTimestampColumn());
+ int count = sorted.size();
+ ArrayList<StreamingMessage> messages = Lists.newArrayList();
+ for (String[] row : sorted) {
+ messages.add((parse(row)));
+ if (messages.size() >= iiDesc.getSliceSize()) {
+ build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable);
+ messages.clear();
+ }
+ }
+
+ if (!messages.isEmpty()) {
+ build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable);
+ }
+
+ reader.close();
+ logger.info("total record count:" + count + " htable:" + segment.getStorageLocationIdentifier());
+ logger.info("stream build finished, htable name:" + segment.getStorageLocationIdentifier());
+ }
+
+ @Test
+ public void test() throws Exception {
+ for (String iiName : II_NAME) {
+ buildII(iiName);
+ IIInstance ii = iiManager.getII(iiName);
+ if (ii.getStatus() != RealizationStatusEnum.READY) {
+ ii.setStatus(RealizationStatusEnum.READY);
+ iiManager.updateII(ii);
+ }
+ }
+ }
+
+ private void build(SliceBuilder sliceBuilder, StreamingBatch batch, HTableInterface htable) throws IOException {
+ final Slice slice = sliceBuilder.buildSlice(batch);
+ try {
+ loadToHBase(htable, slice, new IIKeyValueCodec(slice.getInfo()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void loadToHBase(HTableInterface hTable, Slice slice, IIKeyValueCodec codec) throws IOException {
+ List<Put> data = Lists.newArrayList();
+ for (IIRow row : codec.encodeKeyValue(slice)) {
+ final byte[] key = row.getKey().get();
+ final byte[] value = row.getValue().get();
+ Put put = new Put(key);
+ put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_QUALIFIER_BYTES, value);
+ final ImmutableBytesWritable dictionary = row.getDictionary();
+ final byte[] dictBytes = dictionary.get();
+ if (dictionary.getOffset() == 0 && dictionary.getLength() == dictBytes.length) {
+ put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_DICTIONARY_BYTES, dictBytes);
+ } else {
+ throw new RuntimeException("dict offset should be 0, and dict length should be " + dictBytes.length + " but they are" + dictionary.getOffset() + " " + dictionary.getLength());
+ }
+ data.add(put);
+ }
+ hTable.put(data);
+ //omit hTable.flushCommits(), because htable is auto flush
+ }
+
+ private StreamingMessage parse(String[] row) {
+ return new StreamingMessage(Lists.newArrayList(row), System.currentTimeMillis(), System.currentTimeMillis(), Collections.<String, Object>emptyMap());
+ }
+
+ private List<String[]> getSortedRows(HiveTableReader reader, final int tsCol) throws IOException {
+ List<String[]> unsorted = Lists.newArrayList();
+ while (reader.next()) {
+ unsorted.add(reader.getRow());
+ }
+ Collections.sort(unsorted, new Comparator<String[]>() {
+ @Override
+ public int compare(String[] o1, String[] o2) {
+ long t1 = DateFormat.stringToMillis(o1[tsCol]);
+ long t2 = DateFormat.stringToMillis(o2[tsCol]);
+ return Long.compare(t1, t2);
+ }
+ });
+ return unsorted;
+ }
+
+ private static int cleanupOldStorage() throws Exception {
+ String[] args = { "--delete", "true" };
+
+ int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
+ return exitCode;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/dev-support/test_all_against_hdp_2_2_4_2_2.sh
----------------------------------------------------------------------
diff --git a/dev-support/test_all_against_hdp_2_2_4_2_2.sh b/dev-support/test_all_against_hdp_2_2_4_2_2.sh
index b2f5ce0..e07facb 100644
--- a/dev-support/test_all_against_hdp_2_2_4_2_2.sh
+++ b/dev-support/test_all_against_hdp_2_2_4_2_2.sh
@@ -22,4 +22,7 @@ cd ${dir}
cd ..
mvn clean install -DskipTests | tee mci.log
+mvn test -Dhdp.version=2.2.4.2-2 -Dtest=org.apache.kylin.job.BuildCubeWithEngineTest -DfailIfNoTests=false -P sandbox | tee BuildCubeWithEngineTest.log
+mvn test -Dhdp.version=2.2.4.2-2 -Dtest=org.apache.kylin.job.BuildIIWithStreamTest -DfailIfNoTests=false -P sandbox | tee BuildIIWithStreamTest.log
+mvn test -Dhdp.version=2.2.4.2-2 -Dtest=org.apache.kylin.job.BuildCubeWithStreamTest -DfailIfNoTests=false -P sandbox | tee BuildCubeWithStreamTest.log
mvn verify -Dhdp.version=2.2.4.2-2 -fae -P sandbox | tee mvnverify.log
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/827a995d/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 747e7ee..ccd5a92 100644
--- a/pom.xml
+++ b/pom.xml
@@ -457,11 +457,11 @@
<artifactId>curator-recipes</artifactId>
<version>${curator.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- <version>${apache-httpclient.version}</version>
- </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${apache-httpclient.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>
@@ -647,6 +647,10 @@
<reportsDirectory>${project.basedir}/../target/surefire-reports</reportsDirectory>
<excludes>
<exclude>**/IT*.java</exclude>
+
+ <!--Build cube/II need to be run separately-->
+ <exclude>**/BuildCube*.java</exclude>
+ <exclude>**/BuildII*.java</exclude>
</excludes>
<systemProperties>
<property>
@@ -674,6 +678,7 @@
<excludes>
<exclude>**/*$*</exclude>
<exclude>**/ITKafka*.java</exclude>
+
</excludes>
<systemProperties>
<property>
@@ -683,45 +688,6 @@
</systemProperties>
</configuration>
</plugin>
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>exec-maven-plugin</artifactId>
- <executions>
- <execution>
- <id>build_cube_with_engine</id>
- <goals>
- <goal>java</goal>
- </goals>
- <phase>pre-integration-test</phase>
- <configuration>
- <classpathScope>test</classpathScope>
- <mainClass>org.apache.kylin.job.BuildCubeWithEngine</mainClass>
- </configuration>
- </execution>
- <execution>
- <id>build_cube_with_stream</id>
- <goals>
- <goal>java</goal>
- </goals>
- <phase>pre-integration-test</phase>
- <configuration>
- <classpathScope>test</classpathScope>
- <mainClass>org.apache.kylin.job.BuildCubeWithStream</mainClass>
- </configuration>
- </execution>
- <execution>
- <id>build_ii_with_stream</id>
- <goals>
- <goal>java</goal>
- </goals>
- <phase>pre-integration-test</phase>
- <configuration>
- <classpathScope>test</classpathScope>
- <mainClass>org.apache.kylin.job.BuildIIWithStream</mainClass>
- </configuration>
- </execution>
- </executions>
- </plugin>
</plugins>
</build>
</profile>
@@ -768,9 +734,20 @@
<configuration>
<reportsDirectory>${project.basedir}/../target/surefire-reports</reportsDirectory>
<excludes>
- <excludes>
- <exclude>**/IT*.java</exclude>
- </excludes>
+ <exclude>**/BuildCubeWithEngineTest.java</exclude>
+ <exclude>**/BuildIIWithEngineTest.java</exclude>
+ <exclude>**/SampleCubeSetupTest.java</exclude>
+ <exclude>**/KylinQueryTest.java</exclude>
+ <exclude>**/SnapshotManagerTest.java</exclude>
+ <exclude>**/HiveTableReaderTest.java</exclude>
+ <exclude>**/TableControllerTest.java</exclude>
+
+ <!--test case under server folder are problematic when tested against minicluster-->
+ <exclude>**/ServiceTestBase.java</exclude>
+ <exclude>**/JDBCDriverTest.java</exclude>
+ <exclude>%regex[.*ControllerTest.*]</exclude>
+ <exclude>%regex[.*ServiceTest.*]</exclude>
+
</excludes>
<systemProperties>
<property>
@@ -864,7 +841,7 @@
<exclude>webapp/bower.json</exclude>
<exclude>webapp/grunt.json</exclude>
<exclude>webapp/package.json</exclude>
-
+
<!-- jdbc log -->
<exclude>jdbc/kylin_jdbc.log*</exclude>
<!-- server log -->
[2/2] kylin git commit: minor: wrap exceptions in Observer with
IOException
Posted by ma...@apache.org.
minor: wrap exceptions in Observer with IOException
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/db87d462
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/db87d462
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/db87d462
Branch: refs/heads/2.0-rc
Commit: db87d46293ebe4c47284be02b09315a76e65edce
Parents: a118a27
Author: honma <ho...@ebay.com>
Authored: Fri Jan 22 11:17:21 2016 +0800
Committer: honma <ho...@ebay.com>
Committed: Fri Jan 22 11:19:33 2016 +0800
----------------------------------------------------------------------
.../observer/ObserverAggregationCache.java | 23 ++++++++++++--------
1 file changed, 14 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/db87d462/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregationCache.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregationCache.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregationCache.java
index b7b12d1..d03fe16 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregationCache.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregationCache.java
@@ -31,8 +31,8 @@ import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.Type;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.kylin.metadata.measure.MeasureAggregator;
-import org.apache.kylin.storage.hbase.common.coprocessor.AggregationCache;
import org.apache.kylin.storage.hbase.common.coprocessor.AggrKey;
+import org.apache.kylin.storage.hbase.common.coprocessor.AggregationCache;
/**
* @author yangli9
@@ -67,15 +67,20 @@ public class ObserverAggregationCache extends AggregationCache {
@Override
public boolean next(List<Cell> results) throws IOException {
- // AggregateRegionObserver.LOG.info("Kylin Scanner next()");
- boolean hasMore = false;
- if (iterator.hasNext()) {
- Entry<AggrKey, MeasureAggregator[]> entry = iterator.next();
- makeCells(entry, results);
- hasMore = iterator.hasNext();
+ try {
+ // AggregateRegionObserver.LOG.info("Kylin Scanner next()");
+ boolean hasMore = false;
+ if (iterator.hasNext()) {
+ Entry<AggrKey, MeasureAggregator[]> entry = iterator.next();
+ makeCells(entry, results);
+ hasMore = iterator.hasNext();
+ }
+ // AggregateRegionObserver.LOG.info("Kylin Scanner next() done");
+
+ return hasMore;
+ } catch (Exception e) {
+ throw new IOException("Error when calling next", e);
}
- // AggregateRegionObserver.LOG.info("Kylin Scanner next() done");
- return hasMore;
}
private void makeCells(Entry<AggrKey, MeasureAggregator[]> entry, List<Cell> results) {