@Test public void testSubmitSegmentBuildJob() throws IOException { CubeSegment cubeSegment1 = stubCubSegment(SegmentStatusEnum.NEW, 100L, 200L); CubeSegment cubeSegment2 = stubCubSegment(SegmentStatusEnum.NEW, 1559390400000L, 1559394000000L); CubeInstance cubeInstance = stubCubeInstance(cubeSegment1); @SuppressWarnings("unchecked") Iterator<CubeSegment> cubeSegmentIterable = mock(Iterator.class); when(cubeSegmentIterable.hasNext()).thenReturn(true, false); @SuppressWarnings("unchecked") Segments<CubeSegment> segmentSegments = mock(Segments.class, RETURNS_DEEP_STUBS); when(cubeSegmentIterable.next()).thenReturn(cubeSegment1, cubeSegment2); when(segmentSegments.iterator()).thenReturn(cubeSegmentIterable); when(cubeInstance.getSegments()).thenReturn(segmentSegments); config = stubKylinConfig(); when(cubeInstance.getConfig()).thenReturn(config); cubeManager = stubCubeManager(cubeInstance, false); Map<String, CubingJob> cubingJobMap = new HashMap<>(); cubingJobMap.put(mockBuildJob1, stubCubingJob(ExecutableState.SUCCEED)); cubingJobMap.put(mockBuildJob2, stubCubingJob(ExecutableState.DISCARDED)); cubingJobMap.put(mockBuildJob3, stubCubingJob(ExecutableState.DISCARDED)); cubingJobMap.put(mockBuildJob4, stubCubingJob(ExecutableState.ERROR)); executableManager = stubExecutableManager(cubingJobMap); streamingCoordinator = stubStreamingCoordinator(config, cubeManager, executableManager); clusterManager = stubReceiverClusterManager(streamingCoordinator); when(streamingCoordinator.getClusterManager()).thenReturn(clusterManager); when(cubeManager.appendSegment(any(CubeInstance.class), any(SegmentRange.TSRange.class))) .thenReturn(cubeSegment1, cubeSegment2); BuildJobSubmitter buildJobSubmitter = new BuildJobSubmitter(streamingCoordinator); buildJobSubmitter = spy(buildJobSubmitter); DefaultChainedExecutable cubingJob = mock(DefaultChainedExecutable.class); when(cubingJob.getId()).thenReturn(mockBuildJob4); doReturn(cubingJob).when(buildJobSubmitter).getStreamingCubingJob(any(CubeSegment.class)); buildJobSubmitter.restore(); boolean submitSuccess = buildJobSubmitter.submitSegmentBuildJob(cubeName1, segment1); assertTrue(submitSuccess); }
@NotAtomicIdempotent boolean submitSegmentBuildJob(String cubeName, String segmentName) { logger.info("Try submit streaming segment build job, cube:{} segment:{}", cubeName, segmentName); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); try { CubeSegment newSeg = null; Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentName); boolean segmentExists = false; for (CubeSegment segment : cubeInstance.getSegments()) { SegmentRange.TSRange tsRange = segment.getTSRange(); if (tsRange.start.v.equals(segmentRange.getFirst()) && segmentRange.getSecond().equals(tsRange.end.v)) { segmentExists = true; newSeg = segment; } } if (segmentExists) { logger.warn("Segment {} exists, it will be forced deleted.", segmentName); coordinator.getCubeManager().updateCubeDropSegments(cubeInstance, newSeg); } logger.debug("Create segment for {} {} .", cubeName, segmentName); newSeg = coordinator.getCubeManager().appendSegment(cubeInstance, new SegmentRange.TSRange(segmentRange.getFirst(), segmentRange.getSecond())); DefaultChainedExecutable executable = getStreamingCubingJob(newSeg); coordinator.getExecutableManager().addJob(executable); String jobId = executable.getId(); newSeg.setLastBuildJobID(jobId); SegmentJobBuildInfo segmentJobBuildInfo = new SegmentJobBuildInfo(cubeName, segmentName, jobId); addToJobTrackList(segmentJobBuildInfo); SegmentBuildState.BuildState state = new SegmentBuildState.BuildState(); state.setBuildStartTime(System.currentTimeMillis()); state.setState(SegmentBuildState.BuildState.State.BUILDING); state.setJobId(jobId); logger.debug("Commit building job {} for {} {} .", jobId, cubeName, segmentName); coordinator.getStreamMetadataStore().updateSegmentBuildState(cubeName, segmentName, state); return true; } catch (Exception e) { logger.error("Streaming job submit fail, cubeName:" + cubeName + " segment:" + segmentName, e); return false; } }
BuildJobSubmitter implements Runnable { @NotAtomicIdempotent boolean submitSegmentBuildJob(String cubeName, String segmentName) { logger.info("Try submit streaming segment build job, cube:{} segment:{}", cubeName, segmentName); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); try { CubeSegment newSeg = null; Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentName); boolean segmentExists = false; for (CubeSegment segment : cubeInstance.getSegments()) { SegmentRange.TSRange tsRange = segment.getTSRange(); if (tsRange.start.v.equals(segmentRange.getFirst()) && segmentRange.getSecond().equals(tsRange.end.v)) { segmentExists = true; newSeg = segment; } } if (segmentExists) { logger.warn("Segment {} exists, it will be forced deleted.", segmentName); coordinator.getCubeManager().updateCubeDropSegments(cubeInstance, newSeg); } logger.debug("Create segment for {} {} .", cubeName, segmentName); newSeg = coordinator.getCubeManager().appendSegment(cubeInstance, new SegmentRange.TSRange(segmentRange.getFirst(), segmentRange.getSecond())); DefaultChainedExecutable executable = getStreamingCubingJob(newSeg); coordinator.getExecutableManager().addJob(executable); String jobId = executable.getId(); newSeg.setLastBuildJobID(jobId); SegmentJobBuildInfo segmentJobBuildInfo = new SegmentJobBuildInfo(cubeName, segmentName, jobId); addToJobTrackList(segmentJobBuildInfo); SegmentBuildState.BuildState state = new SegmentBuildState.BuildState(); state.setBuildStartTime(System.currentTimeMillis()); state.setState(SegmentBuildState.BuildState.State.BUILDING); state.setJobId(jobId); logger.debug("Commit building job {} for {} {} .", jobId, cubeName, segmentName); coordinator.getStreamMetadataStore().updateSegmentBuildState(cubeName, segmentName, state); return true; } catch (Exception e) { logger.error("Streaming job submit fail, cubeName:" + cubeName + " segment:" + segmentName, e); return false; } } }
BuildJobSubmitter implements Runnable { @NotAtomicIdempotent boolean submitSegmentBuildJob(String cubeName, String segmentName) { logger.info("Try submit streaming segment build job, cube:{} segment:{}", cubeName, segmentName); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); try { CubeSegment newSeg = null; Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentName); boolean segmentExists = false; for (CubeSegment segment : cubeInstance.getSegments()) { SegmentRange.TSRange tsRange = segment.getTSRange(); if (tsRange.start.v.equals(segmentRange.getFirst()) && segmentRange.getSecond().equals(tsRange.end.v)) { segmentExists = true; newSeg = segment; } } if (segmentExists) { logger.warn("Segment {} exists, it will be forced deleted.", segmentName); coordinator.getCubeManager().updateCubeDropSegments(cubeInstance, newSeg); } logger.debug("Create segment for {} {} .", cubeName, segmentName); newSeg = coordinator.getCubeManager().appendSegment(cubeInstance, new SegmentRange.TSRange(segmentRange.getFirst(), segmentRange.getSecond())); DefaultChainedExecutable executable = getStreamingCubingJob(newSeg); coordinator.getExecutableManager().addJob(executable); String jobId = executable.getId(); newSeg.setLastBuildJobID(jobId); SegmentJobBuildInfo segmentJobBuildInfo = new SegmentJobBuildInfo(cubeName, segmentName, jobId); addToJobTrackList(segmentJobBuildInfo); SegmentBuildState.BuildState state = new SegmentBuildState.BuildState(); state.setBuildStartTime(System.currentTimeMillis()); state.setState(SegmentBuildState.BuildState.State.BUILDING); state.setJobId(jobId); logger.debug("Commit building job {} for {} {} .", jobId, cubeName, segmentName); coordinator.getStreamMetadataStore().updateSegmentBuildState(cubeName, segmentName, state); return true; } catch (Exception e) { logger.error("Streaming job submit fail, cubeName:" + cubeName + " segment:" + segmentName, e); return false; } } BuildJobSubmitter(StreamingCoordinator coordinator); }
BuildJobSubmitter implements Runnable { @NotAtomicIdempotent boolean submitSegmentBuildJob(String cubeName, String segmentName) { logger.info("Try submit streaming segment build job, cube:{} segment:{}", cubeName, segmentName); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); try { CubeSegment newSeg = null; Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentName); boolean segmentExists = false; for (CubeSegment segment : cubeInstance.getSegments()) { SegmentRange.TSRange tsRange = segment.getTSRange(); if (tsRange.start.v.equals(segmentRange.getFirst()) && segmentRange.getSecond().equals(tsRange.end.v)) { segmentExists = true; newSeg = segment; } } if (segmentExists) { logger.warn("Segment {} exists, it will be forced deleted.", segmentName); coordinator.getCubeManager().updateCubeDropSegments(cubeInstance, newSeg); } logger.debug("Create segment for {} {} .", cubeName, segmentName); newSeg = coordinator.getCubeManager().appendSegment(cubeInstance, new SegmentRange.TSRange(segmentRange.getFirst(), segmentRange.getSecond())); DefaultChainedExecutable executable = getStreamingCubingJob(newSeg); coordinator.getExecutableManager().addJob(executable); String jobId = executable.getId(); newSeg.setLastBuildJobID(jobId); SegmentJobBuildInfo segmentJobBuildInfo = new SegmentJobBuildInfo(cubeName, segmentName, jobId); addToJobTrackList(segmentJobBuildInfo); SegmentBuildState.BuildState state = new SegmentBuildState.BuildState(); state.setBuildStartTime(System.currentTimeMillis()); state.setState(SegmentBuildState.BuildState.State.BUILDING); state.setJobId(jobId); logger.debug("Commit building job {} for {} {} .", jobId, cubeName, segmentName); coordinator.getStreamMetadataStore().updateSegmentBuildState(cubeName, segmentName, state); return true; } catch (Exception e) { logger.error("Streaming job submit fail, cubeName:" + cubeName + " segment:" + segmentName, e); return false; } } BuildJobSubmitter(StreamingCoordinator coordinator); @Override void run(); Set<String> getCubeCheckList(); DefaultChainedExecutable getStreamingCubingJob(CubeSegment segment); }
BuildJobSubmitter implements Runnable { @NotAtomicIdempotent boolean submitSegmentBuildJob(String cubeName, String segmentName) { logger.info("Try submit streaming segment build job, cube:{} segment:{}", cubeName, segmentName); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); try { CubeSegment newSeg = null; Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentName); boolean segmentExists = false; for (CubeSegment segment : cubeInstance.getSegments()) { SegmentRange.TSRange tsRange = segment.getTSRange(); if (tsRange.start.v.equals(segmentRange.getFirst()) && segmentRange.getSecond().equals(tsRange.end.v)) { segmentExists = true; newSeg = segment; } } if (segmentExists) { logger.warn("Segment {} exists, it will be forced deleted.", segmentName); coordinator.getCubeManager().updateCubeDropSegments(cubeInstance, newSeg); } logger.debug("Create segment for {} {} .", cubeName, segmentName); newSeg = coordinator.getCubeManager().appendSegment(cubeInstance, new SegmentRange.TSRange(segmentRange.getFirst(), segmentRange.getSecond())); DefaultChainedExecutable executable = getStreamingCubingJob(newSeg); coordinator.getExecutableManager().addJob(executable); String jobId = executable.getId(); newSeg.setLastBuildJobID(jobId); SegmentJobBuildInfo segmentJobBuildInfo = new SegmentJobBuildInfo(cubeName, segmentName, jobId); addToJobTrackList(segmentJobBuildInfo); SegmentBuildState.BuildState state = new SegmentBuildState.BuildState(); state.setBuildStartTime(System.currentTimeMillis()); state.setState(SegmentBuildState.BuildState.State.BUILDING); state.setJobId(jobId); logger.debug("Commit building job {} for {} {} .", jobId, cubeName, segmentName); coordinator.getStreamMetadataStore().updateSegmentBuildState(cubeName, segmentName, state); return true; } catch (Exception e) { logger.error("Streaming job submit fail, cubeName:" + cubeName + " segment:" + segmentName, e); return false; } } BuildJobSubmitter(StreamingCoordinator coordinator); @Override void run(); Set<String> getCubeCheckList(); DefaultChainedExecutable getStreamingCubingJob(CubeSegment segment);  }