@Test public void initAssignTest() { List<ReplicaSet> rsList = Lists.newArrayList(); ReplicaSet rs1 = new ReplicaSet(); rs1.setReplicaSetID(0); ReplicaSet rs2 = new ReplicaSet(); rs2.setReplicaSetID(1); ReplicaSet rs3 = new ReplicaSet(); rs3.setReplicaSetID(2); rsList.add(rs1); rsList.add(rs2); rsList.add(rs3); List<CubeAssignment> existingAssignments = Lists.newArrayList(); Map<Integer, List<Partition>> cube1Assignment = Maps.newHashMap(); cube1Assignment.put(0, Arrays.asList(new Partition(0), new Partition(1), new Partition(2))); CubeAssignment assignment1 = new CubeAssignment("cube1", cube1Assignment); Map<Integer, List<Partition>> cube2Assignment = Maps.newHashMap(); cube2Assignment.put(0, Arrays.asList(new Partition(0), new Partition(1))); cube2Assignment.put(1, Arrays.asList(new Partition(2))); CubeAssignment assignment2 = new CubeAssignment("cube2", cube2Assignment); Map<Integer, List<Partition>> cube3Assignment = Maps.newHashMap(); cube3Assignment.put(2, Arrays.asList(new Partition(0), new Partition(1), new Partition(2))); CubeAssignment assignment3 = new CubeAssignment("cube3", cube3Assignment); Map<Integer, List<Partition>> cube4Assignment = Maps.newHashMap(); cube4Assignment.put(0, Arrays.asList(new Partition(0), new Partition(1), new Partition(2))); CubeAssignment assignment4 = new CubeAssignment("cube4", cube4Assignment); Map<Integer, List<Partition>> cube5Assignment = Maps.newHashMap(); cube5Assignment.put(2, Arrays.asList(new Partition(0), new Partition(1), new Partition(2))); CubeAssignment assignment5 = new CubeAssignment("cube5", cube5Assignment); existingAssignments.add(assignment1); existingAssignments.add(assignment2); existingAssignments.add(assignment3); existingAssignments.add(assignment4); existingAssignments.add(assignment5); StreamingCubeInfo cube5 = new StreamingCubeInfo("cube5", new StreamingTableSourceInfo(Arrays.asList( new Partition(0), new Partition(1))), 2); CubeAssignment cubeAssignment = assigner.assign(cube5, rsList, existingAssignments); assertEquals(cubeAssignment.getAssignments().size(), 2); assertTrue(cubeAssignment.getReplicaSetIDs().contains(1)); assertTrue(cubeAssignment.getReplicaSetIDs().contains(2)); cube5 = new StreamingCubeInfo("cube5", new StreamingTableSourceInfo(Arrays.asList(new Partition(0), new Partition(1))), 3); cubeAssignment = assigner.assign(cube5, rsList, existingAssignments); assertEquals(cubeAssignment.getAssignments().size(), 2); assertTrue(cubeAssignment.getReplicaSetIDs().contains(1)); assertTrue(cubeAssignment.getReplicaSetIDs().contains(2)); cube5 = new StreamingCubeInfo("cube5", new StreamingTableSourceInfo(Arrays.asList(new Partition(0), new Partition(1), new Partition(2))), 5); cubeAssignment = assigner.assign(cube5, rsList, existingAssignments); assertEquals(cubeAssignment.getAssignments().size(), 3); }
@Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { final Map<Integer, Integer> replicaSetTaskNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> rsIds = cubeAssignment.getReplicaSetIDs(); for (Integer rsId : rsIds) { Integer taskNum = replicaSetTaskNumMap.get(rsId); if (taskNum != null) { taskNum = taskNum + 1; } else { taskNum = 1; } replicaSetTaskNumMap.put(rsId, taskNum); } } Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer value1 = replicaSetTaskNumMap.get(o1.getReplicaSetID()); Integer value2 = replicaSetTaskNumMap.get(o2.getReplicaSetID()); int taskNum1 = value1 == null ? 0 : value1; int taskNum2 = value2 == null ? 0 : value2; return taskNum1 - taskNum2; } }); List<List<Partition>> taskPartitions = splitCubeConsumeTasks(cube, replicaSets.size()); Map<Integer, List<Partition>> assignment = Maps.newHashMap(); for (int i = 0; i < taskPartitions.size(); i++) { assignment.put(replicaSets.get(i).getReplicaSetID(), taskPartitions.get(i)); } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignment); return cubeAssignment; }
DefaultAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { final Map<Integer, Integer> replicaSetTaskNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> rsIds = cubeAssignment.getReplicaSetIDs(); for (Integer rsId : rsIds) { Integer taskNum = replicaSetTaskNumMap.get(rsId); if (taskNum != null) { taskNum = taskNum + 1; } else { taskNum = 1; } replicaSetTaskNumMap.put(rsId, taskNum); } } Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer value1 = replicaSetTaskNumMap.get(o1.getReplicaSetID()); Integer value2 = replicaSetTaskNumMap.get(o2.getReplicaSetID()); int taskNum1 = value1 == null ? 0 : value1; int taskNum2 = value2 == null ? 0 : value2; return taskNum1 - taskNum2; } }); List<List<Partition>> taskPartitions = splitCubeConsumeTasks(cube, replicaSets.size()); Map<Integer, List<Partition>> assignment = Maps.newHashMap(); for (int i = 0; i < taskPartitions.size(); i++) { assignment.put(replicaSets.get(i).getReplicaSetID(), taskPartitions.get(i)); } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignment); return cubeAssignment; } }
DefaultAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { final Map<Integer, Integer> replicaSetTaskNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> rsIds = cubeAssignment.getReplicaSetIDs(); for (Integer rsId : rsIds) { Integer taskNum = replicaSetTaskNumMap.get(rsId); if (taskNum != null) { taskNum = taskNum + 1; } else { taskNum = 1; } replicaSetTaskNumMap.put(rsId, taskNum); } } Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer value1 = replicaSetTaskNumMap.get(o1.getReplicaSetID()); Integer value2 = replicaSetTaskNumMap.get(o2.getReplicaSetID()); int taskNum1 = value1 == null ? 0 : value1; int taskNum2 = value2 == null ? 0 : value2; return taskNum1 - taskNum2; } }); List<List<Partition>> taskPartitions = splitCubeConsumeTasks(cube, replicaSets.size()); Map<Integer, List<Partition>> assignment = Maps.newHashMap(); for (int i = 0; i < taskPartitions.size(); i++) { assignment.put(replicaSets.get(i).getReplicaSetID(), taskPartitions.get(i)); } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignment); return cubeAssignment; }  }
DefaultAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { final Map<Integer, Integer> replicaSetTaskNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> rsIds = cubeAssignment.getReplicaSetIDs(); for (Integer rsId : rsIds) { Integer taskNum = replicaSetTaskNumMap.get(rsId); if (taskNum != null) { taskNum = taskNum + 1; } else { taskNum = 1; } replicaSetTaskNumMap.put(rsId, taskNum); } } Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer value1 = replicaSetTaskNumMap.get(o1.getReplicaSetID()); Integer value2 = replicaSetTaskNumMap.get(o2.getReplicaSetID()); int taskNum1 = value1 == null ? 0 : value1; int taskNum2 = value2 == null ? 0 : value2; return taskNum1 - taskNum2; } }); List<List<Partition>> taskPartitions = splitCubeConsumeTasks(cube, replicaSets.size()); Map<Integer, List<Partition>> assignment = Maps.newHashMap(); for (int i = 0; i < taskPartitions.size(); i++) { assignment.put(replicaSets.get(i).getReplicaSetID(), taskPartitions.get(i)); } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignment); 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); }
DefaultAssigner implements Assigner { @Override public CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets, List<CubeAssignment> existingAssignments) { final Map<Integer, Integer> replicaSetTaskNumMap = Maps.newHashMap(); for (CubeAssignment cubeAssignment : existingAssignments) { Set<Integer> rsIds = cubeAssignment.getReplicaSetIDs(); for (Integer rsId : rsIds) { Integer taskNum = replicaSetTaskNumMap.get(rsId); if (taskNum != null) { taskNum = taskNum + 1; } else { taskNum = 1; } replicaSetTaskNumMap.put(rsId, taskNum); } } Collections.sort(replicaSets, new Comparator<ReplicaSet>() { @Override public int compare(ReplicaSet o1, ReplicaSet o2) { Integer value1 = replicaSetTaskNumMap.get(o1.getReplicaSetID()); Integer value2 = replicaSetTaskNumMap.get(o2.getReplicaSetID()); int taskNum1 = value1 == null ? 0 : value1; int taskNum2 = value2 == null ? 0 : value2; return taskNum1 - taskNum2; } }); List<List<Partition>> taskPartitions = splitCubeConsumeTasks(cube, replicaSets.size()); Map<Integer, List<Partition>> assignment = Maps.newHashMap(); for (int i = 0; i < taskPartitions.size(); i++) { assignment.put(replicaSets.get(i).getReplicaSetID(), taskPartitions.get(i)); } CubeAssignment cubeAssignment = new CubeAssignment(cube.getCubeName(), assignment); 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);  }