Add logging info
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/971bf800 Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/971bf800 Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/971bf800 Branch: refs/heads/spark-it Commit: 971bf80009c446a873ddc0927151eb1e68d36e27 Parents: e217ae8 Author: shaofengshi <shaofeng...@apache.org> Authored: Mon Jan 23 11:49:27 2017 +0800 Committer: shaofengshi <shaofeng...@apache.org> Committed: Mon Jan 23 16:23:56 2017 +0800 ---------------------------------------------------------------------- .../org/apache/kylin/source/kafka/job/MergeOffsetStep.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kylin/blob/971bf800/source-kafka/src/main/java/org/apache/kylin/source/kafka/job/MergeOffsetStep.java ---------------------------------------------------------------------- diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/job/MergeOffsetStep.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/job/MergeOffsetStep.java index 9cadd72..914fca2 100644 --- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/job/MergeOffsetStep.java +++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/job/MergeOffsetStep.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.Collections; import java.util.List; +import com.google.common.base.Preconditions; import org.apache.kylin.cube.CubeInstance; import org.apache.kylin.cube.CubeManager; import org.apache.kylin.cube.CubeSegment; @@ -38,6 +39,7 @@ import org.slf4j.LoggerFactory; public class MergeOffsetStep extends AbstractExecutable { private static final Logger logger = LoggerFactory.getLogger(MergeOffsetStep.class); + public MergeOffsetStep() { super(); } @@ -46,12 +48,15 @@ public class MergeOffsetStep extends AbstractExecutable { protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { final CubeManager cubeManager = CubeManager.getInstance(context.getConfig()); final CubeInstance cube = cubeManager.getCube(CubingExecutableUtil.getCubeName(this.getParams())); - final CubeSegment segment = cube.getSegmentById(CubingExecutableUtil.getSegmentId(this.getParams())); + final String segmentId = CubingExecutableUtil.getSegmentId(this.getParams()); + final CubeSegment segment = cube.getSegmentById(segmentId); + Preconditions.checkNotNull(segment, "Cube segment '" + segmentId + "' not found."); List<CubeSegment> mergingSegs = cube.getMergingSegments(segment); - Collections.sort(mergingSegs); + Preconditions.checkArgument(mergingSegs.size() > 0, "Merging segment not exist."); + Collections.sort(mergingSegs); final CubeSegment first = mergingSegs.get(0); final CubeSegment last = mergingSegs.get(mergingSegs.size() - 1);
