@Test public void assignCubeTest() { List<ReplicaSet> replicaSets = Lists.newArrayList(); ReplicaSet rs1 = new ReplicaSet(); rs1.setReplicaSetID(1); ReplicaSet rs2 = new ReplicaSet(); rs2.setReplicaSetID(2); ReplicaSet rs3 = new ReplicaSet(); rs3.setReplicaSetID(3); replicaSets.add(rs1); replicaSets.add(rs2); replicaSets.add(rs3); List<StreamingCubeInfo> cubeInfos = Lists.newArrayList(); createCubeInfoFromMeta("cube5", 7, cubeInfos); StreamingCubeInfo cubeInfo = cubeInfos.get(0); List<CubeAssignment> existingAssignments = Lists.newArrayList(); CubeAssignment cubeAssignment = assigner.assign(cubeInfo, replicaSets, existingAssignments); assertEquals(cubeAssignment.getAssignments().get(rs1.getReplicaSetID()).size(), 3); assertEquals(cubeAssignment.getAssignments().get(rs2.getReplicaSetID()).size(), 2); assertEquals(cubeAssignment.getAssignments().get(rs3.getReplicaSetID()).size(), 2); }
@Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { int existingTotalPartitionNum = 0; int totalPartitionNum = 0; final Map<Integer, Integer> replicaSetPartitionNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> replicaSetIDs = cubeAssignment.getReplicaSetIDs(); for (Integer rsID : replicaSetIDs) { int rsPartitionNum = cubeAssignment.getPartitionsByReplicaSetID(rsID).size(); Integer replicaSetPartitionNum = replicaSetPartitionNumMap.get(rsID); if (replicaSetPartitionNum == null) { replicaSetPartitionNumMap.put(rsID, rsPartitionNum); } else { replicaSetPartitionNumMap.put(rsID, rsPartitionNum + replicaSetPartitionNum); } existingTotalPartitionNum += rsPartitionNum; } } List<Partition> partitionsOfCube = cube.getStreamingTableSourceInfo().getPartitions(); int cubePartitionNum = partitionsOfCube.size(); totalPartitionNum += (existingTotalPartitionNum + cubePartitionNum); int replicaSetsNum = replicaSets.size(); int avgPartitionsPerRS = totalPartitionNum / replicaSetsNum; Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer partitionNum1Obj = replicaSetPartitionNumMap.get(o1.getReplicaSetID()); int partitionNum1 = partitionNum1Obj == null ? 0 : partitionNum1Obj; Integer partitionNum2Obj = replicaSetPartitionNumMap.get(o2.getReplicaSetID()); int partitionNum2 = partitionNum2Obj == null ? 0 : partitionNum2Obj; return partitionNum1 - partitionNum2; } }); int nextAssignPartitionIdx = 0; Map<Integer, List<Partition>> assignments = Maps.newHashMap(); for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Integer partitionNumObj = replicaSetPartitionNumMap.get(replicaSetID); int partitionNum = partitionNumObj == null ? 0 : partitionNumObj; int availableRoom = avgPartitionsPerRS - partitionNum; if (availableRoom <= 0) { continue; } int end = (nextAssignPartitionIdx + availableRoom) < cubePartitionNum ? (nextAssignPartitionIdx + availableRoom) : cubePartitionNum; assignments.put(replicaSetID, Lists.newArrayList(partitionsOfCube.subList(nextAssignPartitionIdx, end))); nextAssignPartitionIdx = end; } if (nextAssignPartitionIdx < cubePartitionNum) { for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Partition part = partitionsOfCube.get(nextAssignPartitionIdx); List<Partition> partitions = assignments.get(replicaSetID); if (partitions == null) { partitions = Lists.newArrayList(); assignments.put(replicaSetID, partitions); } partitions.add(part); nextAssignPartitionIdx++; } } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignments); return cubeAssignment; }
CubePartitionRoundRobinAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { int existingTotalPartitionNum = 0; int totalPartitionNum = 0; final Map<Integer, Integer> replicaSetPartitionNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> replicaSetIDs = cubeAssignment.getReplicaSetIDs(); for (Integer rsID : replicaSetIDs) { int rsPartitionNum = cubeAssignment.getPartitionsByReplicaSetID(rsID).size(); Integer replicaSetPartitionNum = replicaSetPartitionNumMap.get(rsID); if (replicaSetPartitionNum == null) { replicaSetPartitionNumMap.put(rsID, rsPartitionNum); } else { replicaSetPartitionNumMap.put(rsID, rsPartitionNum + replicaSetPartitionNum); } existingTotalPartitionNum += rsPartitionNum; } } List<Partition> partitionsOfCube = cube.getStreamingTableSourceInfo().getPartitions(); int cubePartitionNum = partitionsOfCube.size(); totalPartitionNum += (existingTotalPartitionNum + cubePartitionNum); int replicaSetsNum = replicaSets.size(); int avgPartitionsPerRS = totalPartitionNum / replicaSetsNum; Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer partitionNum1Obj = replicaSetPartitionNumMap.get(o1.getReplicaSetID()); int partitionNum1 = partitionNum1Obj == null ? 0 : partitionNum1Obj; Integer partitionNum2Obj = replicaSetPartitionNumMap.get(o2.getReplicaSetID()); int partitionNum2 = partitionNum2Obj == null ? 0 : partitionNum2Obj; return partitionNum1 - partitionNum2; } }); int nextAssignPartitionIdx = 0; Map<Integer, List<Partition>> assignments = Maps.newHashMap(); for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Integer partitionNumObj = replicaSetPartitionNumMap.get(replicaSetID); int partitionNum = partitionNumObj == null ? 0 : partitionNumObj; int availableRoom = avgPartitionsPerRS - partitionNum; if (availableRoom <= 0) { continue; } int end = (nextAssignPartitionIdx + availableRoom) < cubePartitionNum ? (nextAssignPartitionIdx + availableRoom) : cubePartitionNum; assignments.put(replicaSetID, Lists.newArrayList(partitionsOfCube.subList(nextAssignPartitionIdx, end))); nextAssignPartitionIdx = end; } if (nextAssignPartitionIdx < cubePartitionNum) { for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Partition part = partitionsOfCube.get(nextAssignPartitionIdx); List<Partition> partitions = assignments.get(replicaSetID); if (partitions == null) { partitions = Lists.newArrayList(); assignments.put(replicaSetID, partitions); } partitions.add(part); nextAssignPartitionIdx++; } } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignments); return cubeAssignment; } }
CubePartitionRoundRobinAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { int existingTotalPartitionNum = 0; int totalPartitionNum = 0; final Map<Integer, Integer> replicaSetPartitionNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> replicaSetIDs = cubeAssignment.getReplicaSetIDs(); for (Integer rsID : replicaSetIDs) { int rsPartitionNum = cubeAssignment.getPartitionsByReplicaSetID(rsID).size(); Integer replicaSetPartitionNum = replicaSetPartitionNumMap.get(rsID); if (replicaSetPartitionNum == null) { replicaSetPartitionNumMap.put(rsID, rsPartitionNum); } else { replicaSetPartitionNumMap.put(rsID, rsPartitionNum + replicaSetPartitionNum); } existingTotalPartitionNum += rsPartitionNum; } } List<Partition> partitionsOfCube = cube.getStreamingTableSourceInfo().getPartitions(); int cubePartitionNum = partitionsOfCube.size(); totalPartitionNum += (existingTotalPartitionNum + cubePartitionNum); int replicaSetsNum = replicaSets.size(); int avgPartitionsPerRS = totalPartitionNum / replicaSetsNum; Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer partitionNum1Obj = replicaSetPartitionNumMap.get(o1.getReplicaSetID()); int partitionNum1 = partitionNum1Obj == null ? 0 : partitionNum1Obj; Integer partitionNum2Obj = replicaSetPartitionNumMap.get(o2.getReplicaSetID()); int partitionNum2 = partitionNum2Obj == null ? 0 : partitionNum2Obj; return partitionNum1 - partitionNum2; } }); int nextAssignPartitionIdx = 0; Map<Integer, List<Partition>> assignments = Maps.newHashMap(); for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Integer partitionNumObj = replicaSetPartitionNumMap.get(replicaSetID); int partitionNum = partitionNumObj == null ? 0 : partitionNumObj; int availableRoom = avgPartitionsPerRS - partitionNum; if (availableRoom <= 0) { continue; } int end = (nextAssignPartitionIdx + availableRoom) < cubePartitionNum ? (nextAssignPartitionIdx + availableRoom) : cubePartitionNum; assignments.put(replicaSetID, Lists.newArrayList(partitionsOfCube.subList(nextAssignPartitionIdx, end))); nextAssignPartitionIdx = end; } if (nextAssignPartitionIdx < cubePartitionNum) { for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Partition part = partitionsOfCube.get(nextAssignPartitionIdx); List<Partition> partitions = assignments.get(replicaSetID); if (partitions == null) { partitions = Lists.newArrayList(); assignments.put(replicaSetID, partitions); } partitions.add(part); nextAssignPartitionIdx++; } } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignments); return cubeAssignment; }  }
CubePartitionRoundRobinAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { int existingTotalPartitionNum = 0; int totalPartitionNum = 0; final Map<Integer, Integer> replicaSetPartitionNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> replicaSetIDs = cubeAssignment.getReplicaSetIDs(); for (Integer rsID : replicaSetIDs) { int rsPartitionNum = cubeAssignment.getPartitionsByReplicaSetID(rsID).size(); Integer replicaSetPartitionNum = replicaSetPartitionNumMap.get(rsID); if (replicaSetPartitionNum == null) { replicaSetPartitionNumMap.put(rsID, rsPartitionNum); } else { replicaSetPartitionNumMap.put(rsID, rsPartitionNum + replicaSetPartitionNum); } existingTotalPartitionNum += rsPartitionNum; } } List<Partition> partitionsOfCube = cube.getStreamingTableSourceInfo().getPartitions(); int cubePartitionNum = partitionsOfCube.size(); totalPartitionNum += (existingTotalPartitionNum + cubePartitionNum); int replicaSetsNum = replicaSets.size(); int avgPartitionsPerRS = totalPartitionNum / replicaSetsNum; Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer partitionNum1Obj = replicaSetPartitionNumMap.get(o1.getReplicaSetID()); int partitionNum1 = partitionNum1Obj == null ? 0 : partitionNum1Obj; Integer partitionNum2Obj = replicaSetPartitionNumMap.get(o2.getReplicaSetID()); int partitionNum2 = partitionNum2Obj == null ? 0 : partitionNum2Obj; return partitionNum1 - partitionNum2; } }); int nextAssignPartitionIdx = 0; Map<Integer, List<Partition>> assignments = Maps.newHashMap(); for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Integer partitionNumObj = replicaSetPartitionNumMap.get(replicaSetID); int partitionNum = partitionNumObj == null ? 0 : partitionNumObj; int availableRoom = avgPartitionsPerRS - partitionNum; if (availableRoom <= 0) { continue; } int end = (nextAssignPartitionIdx + availableRoom) < cubePartitionNum ? (nextAssignPartitionIdx + availableRoom) : cubePartitionNum; assignments.put(replicaSetID, Lists.newArrayList(partitionsOfCube.subList(nextAssignPartitionIdx, end))); nextAssignPartitionIdx = end; } if (nextAssignPartitionIdx < cubePartitionNum) { for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Partition part = partitionsOfCube.get(nextAssignPartitionIdx); List<Partition> partitions = assignments.get(replicaSetID); if (partitions == null) { partitions = Lists.newArrayList(); assignments.put(replicaSetID, partitions); } partitions.add(part); nextAssignPartitionIdx++; } } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignments); return cubeAssignment; }  @Override Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets,
                                                                    List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments); @Override CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets,
                                 List<CubeAssignment> existingAssignments); void addToGroupAssignment(Map<String, List<Partition>> groupAssignment, String cubeName, Partition partition); }
CubePartitionRoundRobinAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { int existingTotalPartitionNum = 0; int totalPartitionNum = 0; final Map<Integer, Integer> replicaSetPartitionNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> replicaSetIDs = cubeAssignment.getReplicaSetIDs(); for (Integer rsID : replicaSetIDs) { int rsPartitionNum = cubeAssignment.getPartitionsByReplicaSetID(rsID).size(); Integer replicaSetPartitionNum = replicaSetPartitionNumMap.get(rsID); if (replicaSetPartitionNum == null) { replicaSetPartitionNumMap.put(rsID, rsPartitionNum); } else { replicaSetPartitionNumMap.put(rsID, rsPartitionNum + replicaSetPartitionNum); } existingTotalPartitionNum += rsPartitionNum; } } List<Partition> partitionsOfCube = cube.getStreamingTableSourceInfo().getPartitions(); int cubePartitionNum = partitionsOfCube.size(); totalPartitionNum += (existingTotalPartitionNum + cubePartitionNum); int replicaSetsNum = replicaSets.size(); int avgPartitionsPerRS = totalPartitionNum / replicaSetsNum; Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer partitionNum1Obj = replicaSetPartitionNumMap.get(o1.getReplicaSetID()); int partitionNum1 = partitionNum1Obj == null ? 0 : partitionNum1Obj; Integer partitionNum2Obj = replicaSetPartitionNumMap.get(o2.getReplicaSetID()); int partitionNum2 = partitionNum2Obj == null ? 0 : partitionNum2Obj; return partitionNum1 - partitionNum2; } }); int nextAssignPartitionIdx = 0; Map<Integer, List<Partition>> assignments = Maps.newHashMap(); for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Integer partitionNumObj = replicaSetPartitionNumMap.get(replicaSetID); int partitionNum = partitionNumObj == null ? 0 : partitionNumObj; int availableRoom = avgPartitionsPerRS - partitionNum; if (availableRoom <= 0) { continue; } int end = (nextAssignPartitionIdx + availableRoom) < cubePartitionNum ? (nextAssignPartitionIdx + availableRoom) : cubePartitionNum; assignments.put(replicaSetID, Lists.newArrayList(partitionsOfCube.subList(nextAssignPartitionIdx, end))); nextAssignPartitionIdx = end; } if (nextAssignPartitionIdx < cubePartitionNum) { for (ReplicaSet rs : replicaSets) { if (nextAssignPartitionIdx >= cubePartitionNum) { break; } Integer replicaSetID = rs.getReplicaSetID(); Partition part = partitionsOfCube.get(nextAssignPartitionIdx); List<Partition> partitions = assignments.get(replicaSetID); if (partitions == null) { partitions = Lists.newArrayList(); assignments.put(replicaSetID, partitions); } partitions.add(part); nextAssignPartitionIdx++; } } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignments); return cubeAssignment; }  @Override Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets,
                                                                    List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments); @Override CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets,
                                 List<CubeAssignment> existingAssignments); void addToGroupAssignment(Map<String, List<Partition>> groupAssignment, String cubeName, Partition partition);  }