@Test public void testCheckSegmentBuildJobFromMetadata() { prepareTestCheckSegmentBuildJobFromMetadata(); BuildJobSubmitter buildJobSubmitter = new BuildJobSubmitter(streamingCoordinator); buildJobSubmitter.restore(); List<String> segmentReadyList = buildJobSubmitter.checkSegmentBuildJobFromMetadata(cubeName2); assertEquals(1, segmentReadyList.size()); segmentReadyList = buildJobSubmitter.checkSegmentBuildJobFromMetadata(cubeName3); assertEquals(1, segmentReadyList.size()); }
@NonSideEffect List<String> checkSegmentBuildJobFromMetadata(String cubeName) { List<String> result = Lists.newArrayList(); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); if (isInOptimize(cubeInstance)) { return result; } int allowMaxBuildingSegments = cubeInstance.getConfig().getMaxBuildingSegments(); CubeSegment latestHistoryReadySegment = cubeInstance.getLatestReadySegment(); long minSegmentStart = -1; if (latestHistoryReadySegment != null) { minSegmentStart = latestHistoryReadySegment.getTSRange().end.v; } else { logger.info("there is no ready segments for cube:{}, so only allow 1 segment build concurrently", cubeName); allowMaxBuildingSegments = 1; } CubeAssignment assignments = coordinator.getStreamMetadataStore().getAssignmentsByCube(cubeName); Set<Integer> cubeAssignedReplicaSets = assignments.getReplicaSetIDs(); List<SegmentBuildState> segmentStates = coordinator.getStreamMetadataStore().getSegmentBuildStates(cubeName); int inBuildingSegments = cubeInstance.getBuildingSegments().size(); int leftQuota = allowMaxBuildingSegments - inBuildingSegments; boolean stillQuotaForNewSegment = true; Collections.sort(segmentStates); for (int i = 0; i < segmentStates.size(); i++) { boolean needRebuild = false; if (leftQuota <= 0) { logger.info("No left quota to build segments for cube:{} at {}", cubeName, leftQuota); stillQuotaForNewSegment = false; } SegmentBuildState segmentState = segmentStates.get(i); Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentState.getSegmentName()); if (segmentRange.getFirst() < minSegmentStart) { logger.warn( "The cube segment state is not correct because it belongs to historcial part, cube:{} segment:{}, clear it.", cubeName, segmentState.getSegmentName()); coordinator.getStreamMetadataStore().removeSegmentBuildState(cubeName, segmentState.getSegmentName()); continue; } if (segmentState.isInBuilding()) { needRebuild = checkSegmentBuildingJob(segmentState, cubeName, cubeInstance); if (!needRebuild) continue; } else if (segmentState.isInWaiting()) { } boolean readyToBuild = checkSegmentIsReadyToBuild(segmentStates, i, cubeAssignedReplicaSets); if (!readyToBuild) { logger.debug("Segment {} {} is not ready to submit a building job.", cubeName, segmentState); } else if (stillQuotaForNewSegment || needRebuild) { result.add(segmentState.getSegmentName()); leftQuota--; } } if (logger.isDebugEnabled() && !result.isEmpty()) { logger.debug("{} Candidate segment list to be built : {}.", cubeName, String.join(", ", result)); } return result; }
BuildJobSubmitter implements Runnable { @NonSideEffect List<String> checkSegmentBuildJobFromMetadata(String cubeName) { List<String> result = Lists.newArrayList(); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); if (isInOptimize(cubeInstance)) { return result; } int allowMaxBuildingSegments = cubeInstance.getConfig().getMaxBuildingSegments(); CubeSegment latestHistoryReadySegment = cubeInstance.getLatestReadySegment(); long minSegmentStart = -1; if (latestHistoryReadySegment != null) { minSegmentStart = latestHistoryReadySegment.getTSRange().end.v; } else { logger.info("there is no ready segments for cube:{}, so only allow 1 segment build concurrently", cubeName); allowMaxBuildingSegments = 1; } CubeAssignment assignments = coordinator.getStreamMetadataStore().getAssignmentsByCube(cubeName); Set<Integer> cubeAssignedReplicaSets = assignments.getReplicaSetIDs(); List<SegmentBuildState> segmentStates = coordinator.getStreamMetadataStore().getSegmentBuildStates(cubeName); int inBuildingSegments = cubeInstance.getBuildingSegments().size(); int leftQuota = allowMaxBuildingSegments - inBuildingSegments; boolean stillQuotaForNewSegment = true; Collections.sort(segmentStates); for (int i = 0; i < segmentStates.size(); i++) { boolean needRebuild = false; if (leftQuota <= 0) { logger.info("No left quota to build segments for cube:{} at {}", cubeName, leftQuota); stillQuotaForNewSegment = false; } SegmentBuildState segmentState = segmentStates.get(i); Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentState.getSegmentName()); if (segmentRange.getFirst() < minSegmentStart) { logger.warn( "The cube segment state is not correct because it belongs to historcial part, cube:{} segment:{}, clear it.", cubeName, segmentState.getSegmentName()); coordinator.getStreamMetadataStore().removeSegmentBuildState(cubeName, segmentState.getSegmentName()); continue; } if (segmentState.isInBuilding()) { needRebuild = checkSegmentBuildingJob(segmentState, cubeName, cubeInstance); if (!needRebuild) continue; } else if (segmentState.isInWaiting()) { } boolean readyToBuild = checkSegmentIsReadyToBuild(segmentStates, i, cubeAssignedReplicaSets); if (!readyToBuild) { logger.debug("Segment {} {} is not ready to submit a building job.", cubeName, segmentState); } else if (stillQuotaForNewSegment || needRebuild) { result.add(segmentState.getSegmentName()); leftQuota--; } } if (logger.isDebugEnabled() && !result.isEmpty()) { logger.debug("{} Candidate segment list to be built : {}.", cubeName, String.join(", ", result)); } return result; } }
BuildJobSubmitter implements Runnable { @NonSideEffect List<String> checkSegmentBuildJobFromMetadata(String cubeName) { List<String> result = Lists.newArrayList(); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); if (isInOptimize(cubeInstance)) { return result; } int allowMaxBuildingSegments = cubeInstance.getConfig().getMaxBuildingSegments(); CubeSegment latestHistoryReadySegment = cubeInstance.getLatestReadySegment(); long minSegmentStart = -1; if (latestHistoryReadySegment != null) { minSegmentStart = latestHistoryReadySegment.getTSRange().end.v; } else { logger.info("there is no ready segments for cube:{}, so only allow 1 segment build concurrently", cubeName); allowMaxBuildingSegments = 1; } CubeAssignment assignments = coordinator.getStreamMetadataStore().getAssignmentsByCube(cubeName); Set<Integer> cubeAssignedReplicaSets = assignments.getReplicaSetIDs(); List<SegmentBuildState> segmentStates = coordinator.getStreamMetadataStore().getSegmentBuildStates(cubeName); int inBuildingSegments = cubeInstance.getBuildingSegments().size(); int leftQuota = allowMaxBuildingSegments - inBuildingSegments; boolean stillQuotaForNewSegment = true; Collections.sort(segmentStates); for (int i = 0; i < segmentStates.size(); i++) { boolean needRebuild = false; if (leftQuota <= 0) { logger.info("No left quota to build segments for cube:{} at {}", cubeName, leftQuota); stillQuotaForNewSegment = false; } SegmentBuildState segmentState = segmentStates.get(i); Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentState.getSegmentName()); if (segmentRange.getFirst() < minSegmentStart) { logger.warn( "The cube segment state is not correct because it belongs to historcial part, cube:{} segment:{}, clear it.", cubeName, segmentState.getSegmentName()); coordinator.getStreamMetadataStore().removeSegmentBuildState(cubeName, segmentState.getSegmentName()); continue; } if (segmentState.isInBuilding()) { needRebuild = checkSegmentBuildingJob(segmentState, cubeName, cubeInstance); if (!needRebuild) continue; } else if (segmentState.isInWaiting()) { } boolean readyToBuild = checkSegmentIsReadyToBuild(segmentStates, i, cubeAssignedReplicaSets); if (!readyToBuild) { logger.debug("Segment {} {} is not ready to submit a building job.", cubeName, segmentState); } else if (stillQuotaForNewSegment || needRebuild) { result.add(segmentState.getSegmentName()); leftQuota--; } } if (logger.isDebugEnabled() && !result.isEmpty()) { logger.debug("{} Candidate segment list to be built : {}.", cubeName, String.join(", ", result)); } return result; } BuildJobSubmitter(StreamingCoordinator coordinator); }
BuildJobSubmitter implements Runnable { @NonSideEffect List<String> checkSegmentBuildJobFromMetadata(String cubeName) { List<String> result = Lists.newArrayList(); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); if (isInOptimize(cubeInstance)) { return result; } int allowMaxBuildingSegments = cubeInstance.getConfig().getMaxBuildingSegments(); CubeSegment latestHistoryReadySegment = cubeInstance.getLatestReadySegment(); long minSegmentStart = -1; if (latestHistoryReadySegment != null) { minSegmentStart = latestHistoryReadySegment.getTSRange().end.v; } else { logger.info("there is no ready segments for cube:{}, so only allow 1 segment build concurrently", cubeName); allowMaxBuildingSegments = 1; } CubeAssignment assignments = coordinator.getStreamMetadataStore().getAssignmentsByCube(cubeName); Set<Integer> cubeAssignedReplicaSets = assignments.getReplicaSetIDs(); List<SegmentBuildState> segmentStates = coordinator.getStreamMetadataStore().getSegmentBuildStates(cubeName); int inBuildingSegments = cubeInstance.getBuildingSegments().size(); int leftQuota = allowMaxBuildingSegments - inBuildingSegments; boolean stillQuotaForNewSegment = true; Collections.sort(segmentStates); for (int i = 0; i < segmentStates.size(); i++) { boolean needRebuild = false; if (leftQuota <= 0) { logger.info("No left quota to build segments for cube:{} at {}", cubeName, leftQuota); stillQuotaForNewSegment = false; } SegmentBuildState segmentState = segmentStates.get(i); Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentState.getSegmentName()); if (segmentRange.getFirst() < minSegmentStart) { logger.warn( "The cube segment state is not correct because it belongs to historcial part, cube:{} segment:{}, clear it.", cubeName, segmentState.getSegmentName()); coordinator.getStreamMetadataStore().removeSegmentBuildState(cubeName, segmentState.getSegmentName()); continue; } if (segmentState.isInBuilding()) { needRebuild = checkSegmentBuildingJob(segmentState, cubeName, cubeInstance); if (!needRebuild) continue; } else if (segmentState.isInWaiting()) { } boolean readyToBuild = checkSegmentIsReadyToBuild(segmentStates, i, cubeAssignedReplicaSets); if (!readyToBuild) { logger.debug("Segment {} {} is not ready to submit a building job.", cubeName, segmentState); } else if (stillQuotaForNewSegment || needRebuild) { result.add(segmentState.getSegmentName()); leftQuota--; } } if (logger.isDebugEnabled() && !result.isEmpty()) { logger.debug("{} Candidate segment list to be built : {}.", cubeName, String.join(", ", result)); } return result; } BuildJobSubmitter(StreamingCoordinator coordinator); @Override void run(); Set<String> getCubeCheckList(); DefaultChainedExecutable getStreamingCubingJob(CubeSegment segment); }
BuildJobSubmitter implements Runnable { @NonSideEffect List<String> checkSegmentBuildJobFromMetadata(String cubeName) { List<String> result = Lists.newArrayList(); CubeInstance cubeInstance = coordinator.getCubeManager().getCube(cubeName); if (isInOptimize(cubeInstance)) { return result; } int allowMaxBuildingSegments = cubeInstance.getConfig().getMaxBuildingSegments(); CubeSegment latestHistoryReadySegment = cubeInstance.getLatestReadySegment(); long minSegmentStart = -1; if (latestHistoryReadySegment != null) { minSegmentStart = latestHistoryReadySegment.getTSRange().end.v; } else { logger.info("there is no ready segments for cube:{}, so only allow 1 segment build concurrently", cubeName); allowMaxBuildingSegments = 1; } CubeAssignment assignments = coordinator.getStreamMetadataStore().getAssignmentsByCube(cubeName); Set<Integer> cubeAssignedReplicaSets = assignments.getReplicaSetIDs(); List<SegmentBuildState> segmentStates = coordinator.getStreamMetadataStore().getSegmentBuildStates(cubeName); int inBuildingSegments = cubeInstance.getBuildingSegments().size(); int leftQuota = allowMaxBuildingSegments - inBuildingSegments; boolean stillQuotaForNewSegment = true; Collections.sort(segmentStates); for (int i = 0; i < segmentStates.size(); i++) { boolean needRebuild = false; if (leftQuota <= 0) { logger.info("No left quota to build segments for cube:{} at {}", cubeName, leftQuota); stillQuotaForNewSegment = false; } SegmentBuildState segmentState = segmentStates.get(i); Pair<Long, Long> segmentRange = CubeSegment.parseSegmentName(segmentState.getSegmentName()); if (segmentRange.getFirst() < minSegmentStart) { logger.warn( "The cube segment state is not correct because it belongs to historcial part, cube:{} segment:{}, clear it.", cubeName, segmentState.getSegmentName()); coordinator.getStreamMetadataStore().removeSegmentBuildState(cubeName, segmentState.getSegmentName()); continue; } if (segmentState.isInBuilding()) { needRebuild = checkSegmentBuildingJob(segmentState, cubeName, cubeInstance); if (!needRebuild) continue; } else if (segmentState.isInWaiting()) { } boolean readyToBuild = checkSegmentIsReadyToBuild(segmentStates, i, cubeAssignedReplicaSets); if (!readyToBuild) { logger.debug("Segment {} {} is not ready to submit a building job.", cubeName, segmentState); } else if (stillQuotaForNewSegment || needRebuild) { result.add(segmentState.getSegmentName()); leftQuota--; } } if (logger.isDebugEnabled() && !result.isEmpty()) { logger.debug("{} Candidate segment list to be built : {}.", cubeName, String.join(", ", result)); } return result; } BuildJobSubmitter(StreamingCoordinator coordinator); @Override void run(); Set<String> getCubeCheckList(); DefaultChainedExecutable getStreamingCubingJob(CubeSegment segment);  }