@Test public void testTraceEarliestSegmentBuildJob() { beforeTestTraceEarliestSegmentBuildJob(); BuildJobSubmitter buildJobSubmitter = new BuildJobSubmitter(streamingCoordinator); buildJobSubmitter.restore(); List<SegmentJobBuildInfo> jobList = buildJobSubmitter.traceEarliestSegmentBuildJob(); assertEquals(1, jobList.size()); assertThat(jobList.stream().map(x -> x.jobID).collect(Collectors.toSet()), hasItem(mockBuildJob1)); assertEquals(1, buildJobSubmitter.getCubeCheckList().size()); }
@NonSideEffect List<SegmentJobBuildInfo> traceEarliestSegmentBuildJob() { List<SegmentJobBuildInfo> successJobs = Lists.newArrayList(); for (Map.Entry<String, ConcurrentSkipListSet<SegmentJobBuildInfo>> entry : segmentBuildJobCheckList.entrySet()) { ConcurrentSkipListSet<SegmentJobBuildInfo> buildInfos = entry.getValue(); if (buildInfos.isEmpty()) { logger.trace("Skip {}", entry.getKey()); continue; } SegmentJobBuildInfo segmentBuildJob = buildInfos.first(); logger.debug("Check the cube:{} segment:{} build status.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); try { CubingJob cubingJob = (CubingJob) coordinator.getExecutableManager().getJob(segmentBuildJob.jobID); if (cubingJob == null) { logger.error("Cannot find metadata of current job."); continue; } ExecutableState jobState = cubingJob.getStatus(); logger.debug("Current job state {}", jobState); if (ExecutableState.SUCCEED.equals(jobState)) { CubeManager cubeManager = coordinator.getCubeManager(); CubeInstance cubeInstance = cubeManager.getCube(segmentBuildJob.cubeName).latestCopyForWrite(); CubeSegment cubeSegment = cubeInstance.getSegment(segmentBuildJob.segmentName, null); logger.info("The cube:{} segment:{} is ready to be promoted.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); coordinator.getClusterManager().segmentBuildComplete(cubingJob, cubeInstance, cubeSegment, segmentBuildJob); addToCheckList(cubeInstance.getName()); successJobs.add(segmentBuildJob); } else if (ExecutableState.ERROR.equals(jobState)) { if (segmentBuildJob.retryCnt < 5) { logger.info("Job:{} is error, resume the job.", segmentBuildJob); coordinator.getExecutableManager().resumeJob(segmentBuildJob.jobID); segmentBuildJob.retryCnt++; } else { logger.warn("Job:{} is error, exceed max retry. Kylin admin could resume it or discard it" + "(to let new building job be sumbitted) .", segmentBuildJob); } } } catch (StoreException storeEx) { logger.error("Error when check streaming segment job build state:" + segmentBuildJob, storeEx); throw storeEx; } } return successJobs; }
BuildJobSubmitter implements Runnable { @NonSideEffect List<SegmentJobBuildInfo> traceEarliestSegmentBuildJob() { List<SegmentJobBuildInfo> successJobs = Lists.newArrayList(); for (Map.Entry<String, ConcurrentSkipListSet<SegmentJobBuildInfo>> entry : segmentBuildJobCheckList.entrySet()) { ConcurrentSkipListSet<SegmentJobBuildInfo> buildInfos = entry.getValue(); if (buildInfos.isEmpty()) { logger.trace("Skip {}", entry.getKey()); continue; } SegmentJobBuildInfo segmentBuildJob = buildInfos.first(); logger.debug("Check the cube:{} segment:{} build status.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); try { CubingJob cubingJob = (CubingJob) coordinator.getExecutableManager().getJob(segmentBuildJob.jobID); if (cubingJob == null) { logger.error("Cannot find metadata of current job."); continue; } ExecutableState jobState = cubingJob.getStatus(); logger.debug("Current job state {}", jobState); if (ExecutableState.SUCCEED.equals(jobState)) { CubeManager cubeManager = coordinator.getCubeManager(); CubeInstance cubeInstance = cubeManager.getCube(segmentBuildJob.cubeName).latestCopyForWrite(); CubeSegment cubeSegment = cubeInstance.getSegment(segmentBuildJob.segmentName, null); logger.info("The cube:{} segment:{} is ready to be promoted.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); coordinator.getClusterManager().segmentBuildComplete(cubingJob, cubeInstance, cubeSegment, segmentBuildJob); addToCheckList(cubeInstance.getName()); successJobs.add(segmentBuildJob); } else if (ExecutableState.ERROR.equals(jobState)) { if (segmentBuildJob.retryCnt < 5) { logger.info("Job:{} is error, resume the job.", segmentBuildJob); coordinator.getExecutableManager().resumeJob(segmentBuildJob.jobID); segmentBuildJob.retryCnt++; } else { logger.warn("Job:{} is error, exceed max retry. Kylin admin could resume it or discard it" + "(to let new building job be sumbitted) .", segmentBuildJob); } } } catch (StoreException storeEx) { logger.error("Error when check streaming segment job build state:" + segmentBuildJob, storeEx); throw storeEx; } } return successJobs; } }
BuildJobSubmitter implements Runnable { @NonSideEffect List<SegmentJobBuildInfo> traceEarliestSegmentBuildJob() { List<SegmentJobBuildInfo> successJobs = Lists.newArrayList(); for (Map.Entry<String, ConcurrentSkipListSet<SegmentJobBuildInfo>> entry : segmentBuildJobCheckList.entrySet()) { ConcurrentSkipListSet<SegmentJobBuildInfo> buildInfos = entry.getValue(); if (buildInfos.isEmpty()) { logger.trace("Skip {}", entry.getKey()); continue; } SegmentJobBuildInfo segmentBuildJob = buildInfos.first(); logger.debug("Check the cube:{} segment:{} build status.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); try { CubingJob cubingJob = (CubingJob) coordinator.getExecutableManager().getJob(segmentBuildJob.jobID); if (cubingJob == null) { logger.error("Cannot find metadata of current job."); continue; } ExecutableState jobState = cubingJob.getStatus(); logger.debug("Current job state {}", jobState); if (ExecutableState.SUCCEED.equals(jobState)) { CubeManager cubeManager = coordinator.getCubeManager(); CubeInstance cubeInstance = cubeManager.getCube(segmentBuildJob.cubeName).latestCopyForWrite(); CubeSegment cubeSegment = cubeInstance.getSegment(segmentBuildJob.segmentName, null); logger.info("The cube:{} segment:{} is ready to be promoted.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); coordinator.getClusterManager().segmentBuildComplete(cubingJob, cubeInstance, cubeSegment, segmentBuildJob); addToCheckList(cubeInstance.getName()); successJobs.add(segmentBuildJob); } else if (ExecutableState.ERROR.equals(jobState)) { if (segmentBuildJob.retryCnt < 5) { logger.info("Job:{} is error, resume the job.", segmentBuildJob); coordinator.getExecutableManager().resumeJob(segmentBuildJob.jobID); segmentBuildJob.retryCnt++; } else { logger.warn("Job:{} is error, exceed max retry. Kylin admin could resume it or discard it" + "(to let new building job be sumbitted) .", segmentBuildJob); } } } catch (StoreException storeEx) { logger.error("Error when check streaming segment job build state:" + segmentBuildJob, storeEx); throw storeEx; } } return successJobs; } BuildJobSubmitter(StreamingCoordinator coordinator); }
BuildJobSubmitter implements Runnable { @NonSideEffect List<SegmentJobBuildInfo> traceEarliestSegmentBuildJob() { List<SegmentJobBuildInfo> successJobs = Lists.newArrayList(); for (Map.Entry<String, ConcurrentSkipListSet<SegmentJobBuildInfo>> entry : segmentBuildJobCheckList.entrySet()) { ConcurrentSkipListSet<SegmentJobBuildInfo> buildInfos = entry.getValue(); if (buildInfos.isEmpty()) { logger.trace("Skip {}", entry.getKey()); continue; } SegmentJobBuildInfo segmentBuildJob = buildInfos.first(); logger.debug("Check the cube:{} segment:{} build status.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); try { CubingJob cubingJob = (CubingJob) coordinator.getExecutableManager().getJob(segmentBuildJob.jobID); if (cubingJob == null) { logger.error("Cannot find metadata of current job."); continue; } ExecutableState jobState = cubingJob.getStatus(); logger.debug("Current job state {}", jobState); if (ExecutableState.SUCCEED.equals(jobState)) { CubeManager cubeManager = coordinator.getCubeManager(); CubeInstance cubeInstance = cubeManager.getCube(segmentBuildJob.cubeName).latestCopyForWrite(); CubeSegment cubeSegment = cubeInstance.getSegment(segmentBuildJob.segmentName, null); logger.info("The cube:{} segment:{} is ready to be promoted.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); coordinator.getClusterManager().segmentBuildComplete(cubingJob, cubeInstance, cubeSegment, segmentBuildJob); addToCheckList(cubeInstance.getName()); successJobs.add(segmentBuildJob); } else if (ExecutableState.ERROR.equals(jobState)) { if (segmentBuildJob.retryCnt < 5) { logger.info("Job:{} is error, resume the job.", segmentBuildJob); coordinator.getExecutableManager().resumeJob(segmentBuildJob.jobID); segmentBuildJob.retryCnt++; } else { logger.warn("Job:{} is error, exceed max retry. Kylin admin could resume it or discard it" + "(to let new building job be sumbitted) .", segmentBuildJob); } } } catch (StoreException storeEx) { logger.error("Error when check streaming segment job build state:" + segmentBuildJob, storeEx); throw storeEx; } } return successJobs; } BuildJobSubmitter(StreamingCoordinator coordinator); @Override void run(); Set<String> getCubeCheckList(); DefaultChainedExecutable getStreamingCubingJob(CubeSegment segment); }
BuildJobSubmitter implements Runnable { @NonSideEffect List<SegmentJobBuildInfo> traceEarliestSegmentBuildJob() { List<SegmentJobBuildInfo> successJobs = Lists.newArrayList(); for (Map.Entry<String, ConcurrentSkipListSet<SegmentJobBuildInfo>> entry : segmentBuildJobCheckList.entrySet()) { ConcurrentSkipListSet<SegmentJobBuildInfo> buildInfos = entry.getValue(); if (buildInfos.isEmpty()) { logger.trace("Skip {}", entry.getKey()); continue; } SegmentJobBuildInfo segmentBuildJob = buildInfos.first(); logger.debug("Check the cube:{} segment:{} build status.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); try { CubingJob cubingJob = (CubingJob) coordinator.getExecutableManager().getJob(segmentBuildJob.jobID); if (cubingJob == null) { logger.error("Cannot find metadata of current job."); continue; } ExecutableState jobState = cubingJob.getStatus(); logger.debug("Current job state {}", jobState); if (ExecutableState.SUCCEED.equals(jobState)) { CubeManager cubeManager = coordinator.getCubeManager(); CubeInstance cubeInstance = cubeManager.getCube(segmentBuildJob.cubeName).latestCopyForWrite(); CubeSegment cubeSegment = cubeInstance.getSegment(segmentBuildJob.segmentName, null); logger.info("The cube:{} segment:{} is ready to be promoted.", segmentBuildJob.cubeName, segmentBuildJob.segmentName); coordinator.getClusterManager().segmentBuildComplete(cubingJob, cubeInstance, cubeSegment, segmentBuildJob); addToCheckList(cubeInstance.getName()); successJobs.add(segmentBuildJob); } else if (ExecutableState.ERROR.equals(jobState)) { if (segmentBuildJob.retryCnt < 5) { logger.info("Job:{} is error, resume the job.", segmentBuildJob); coordinator.getExecutableManager().resumeJob(segmentBuildJob.jobID); segmentBuildJob.retryCnt++; } else { logger.warn("Job:{} is error, exceed max retry. Kylin admin could resume it or discard it" + "(to let new building job be sumbitted) .", segmentBuildJob); } } } catch (StoreException storeEx) { logger.error("Error when check streaming segment job build state:" + segmentBuildJob, storeEx); throw storeEx; } } return successJobs; } BuildJobSubmitter(StreamingCoordinator coordinator); @Override void run(); Set<String> getCubeCheckList(); DefaultChainedExecutable getStreamingCubingJob(CubeSegment segment);  }