@Test public void reBalanceTest() { 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 cube1 = new StreamingCubeInfo("cube1", new StreamingTableSourceInfo(Arrays.asList( new Partition(0), new Partition(1), new Partition(2))), 1); StreamingCubeInfo cube2 = new StreamingCubeInfo("cube2", new StreamingTableSourceInfo(Arrays.asList( new Partition(0), new Partition(1), new Partition(2))), 2); StreamingCubeInfo cube3 = new StreamingCubeInfo("cube3", new StreamingTableSourceInfo(Arrays.asList( new Partition(0), new Partition(1), new Partition(2))), 1); StreamingCubeInfo cube4 = new StreamingCubeInfo("cube4", new StreamingTableSourceInfo(Arrays.asList( new Partition(0), new Partition(1), new Partition(2))), 1); StreamingCubeInfo cube5 = new StreamingCubeInfo("cube5", new StreamingTableSourceInfo(Arrays.asList( new Partition(0), new Partition(1), new Partition(2))), 1); List<StreamingCubeInfo> cubes = Arrays.asList(cube1, cube2, cube3, cube4, cube5); Map<Integer, Map<String, List<Partition>>> result = assigner.reBalancePlan(rsList, cubes, existingAssignments); for (Map<String, List<Partition>> rsAssignment : result.values()) { assertEquals(2, rsAssignment.size()); } ReplicaSet rs4 = new ReplicaSet(); rs4.setReplicaSetID(3); ReplicaSet rs5 = new ReplicaSet(); rs5.setReplicaSetID(4); ReplicaSet rs6 = new ReplicaSet(); rs6.setReplicaSetID(5); rsList.add(rs4); rsList.add(rs5); rsList.add(rs6); result = assigner.reBalancePlan(rsList, cubes, existingAssignments); for (Map<String, List<Partition>> rsAssignment : result.values()) { assertEquals(1, rsAssignment.size()); } }
@Override public Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets, List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments) { Map<Integer, Map<String, List<Partition>>> newPlan = Maps.newHashMap(); if (replicaSets == null || cubes == null || cubes.size() == 0 || replicaSets.size() == 0) { return newPlan; } Map<Integer, Map<String, List<Partition>>> existingRSAssignmentsMap = AssignmentUtil .convertCubeAssign2ReplicaSetAssign(existingAssignments); Map<String, CubeAssignment> cubeAssignmentMap = Maps.newHashMap(); for (CubeAssignment existingAssignment : existingAssignments) { cubeAssignmentMap.put(existingAssignment.getCubeName(), existingAssignment); } Set<Integer> currReplicaSetIDs = Sets.newHashSet(); for (ReplicaSet replicaSet : replicaSets) { currReplicaSetIDs.add(replicaSet.getReplicaSetID()); } int totalTasks = 0; Map<String, List<List<Partition>>> requireReassignTaskMap = Maps.newHashMap(); Set<String> partitionChangeCubes = Sets.newHashSet(); for (StreamingCubeInfo cube : cubes) { CubeAssignment existingAssignment = cubeAssignmentMap.get(cube.getCubeName()); int prevPartitionNum = existingAssignment.getPartitionIDs().size(); int currPartitionNum = cube.getStreamingTableSourceInfo().getPartitions().size(); if (prevPartitionNum != currPartitionNum) { List<List<Partition>> cubeConsumeTasks = splitCubeConsumeTasks(cube, replicaSets.size()); requireReassignTaskMap.put(cube.getCubeName(), cubeConsumeTasks); partitionChangeCubes.add(cube.getCubeName()); } totalTasks += cube.getNumOfConsumerTasks(); } int avgTasks = totalTasks / replicaSets.size(); for (Entry<Integer, Map<String, List<Partition>>> rsAssignmentEntry : existingRSAssignmentsMap.entrySet()) { Integer rsId = rsAssignmentEntry.getKey(); Map<String, List<Partition>> rsAssignment = rsAssignmentEntry.getValue(); if (!currReplicaSetIDs.contains(rsId)) { throw new IllegalStateException("current replica sets don't contain rs:" + rsId); } Map<String, List<Partition>> newRsTaskMap = newPlan.get(rsId); if (newRsTaskMap == null) { newRsTaskMap = Maps.newHashMap(); newPlan.put(rsId, newRsTaskMap); } for (Entry<String, List<Partition>> taskEntry : rsAssignment.entrySet()) { String cubeName = taskEntry.getKey(); List<Partition> partitions = taskEntry.getValue(); if (partitionChangeCubes.contains(cubeName)) { continue; } if (newRsTaskMap.size() < avgTasks) { newRsTaskMap.put(cubeName, partitions); } else { List<List<Partition>> cubeTasks = requireReassignTaskMap.get(cubeName); if (cubeTasks == null) { cubeTasks = Lists.newArrayList(); requireReassignTaskMap.put(cubeName, cubeTasks); } cubeTasks.add(partitions); } } } for (Entry<String, List<List<Partition>>> requireReassignTaskEntry : requireReassignTaskMap.entrySet()) { setNewPlanForCube(requireReassignTaskEntry.getKey(), requireReassignTaskEntry.getValue(), replicaSets, newPlan); } return newPlan; }
DefaultAssigner implements Assigner { @Override public Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets, List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments) { Map<Integer, Map<String, List<Partition>>> newPlan = Maps.newHashMap(); if (replicaSets == null || cubes == null || cubes.size() == 0 || replicaSets.size() == 0) { return newPlan; } Map<Integer, Map<String, List<Partition>>> existingRSAssignmentsMap = AssignmentUtil .convertCubeAssign2ReplicaSetAssign(existingAssignments); Map<String, CubeAssignment> cubeAssignmentMap = Maps.newHashMap(); for (CubeAssignment existingAssignment : existingAssignments) { cubeAssignmentMap.put(existingAssignment.getCubeName(), existingAssignment); } Set<Integer> currReplicaSetIDs = Sets.newHashSet(); for (ReplicaSet replicaSet : replicaSets) { currReplicaSetIDs.add(replicaSet.getReplicaSetID()); } int totalTasks = 0; Map<String, List<List<Partition>>> requireReassignTaskMap = Maps.newHashMap(); Set<String> partitionChangeCubes = Sets.newHashSet(); for (StreamingCubeInfo cube : cubes) { CubeAssignment existingAssignment = cubeAssignmentMap.get(cube.getCubeName()); int prevPartitionNum = existingAssignment.getPartitionIDs().size(); int currPartitionNum = cube.getStreamingTableSourceInfo().getPartitions().size(); if (prevPartitionNum != currPartitionNum) { List<List<Partition>> cubeConsumeTasks = splitCubeConsumeTasks(cube, replicaSets.size()); requireReassignTaskMap.put(cube.getCubeName(), cubeConsumeTasks); partitionChangeCubes.add(cube.getCubeName()); } totalTasks += cube.getNumOfConsumerTasks(); } int avgTasks = totalTasks / replicaSets.size(); for (Entry<Integer, Map<String, List<Partition>>> rsAssignmentEntry : existingRSAssignmentsMap.entrySet()) { Integer rsId = rsAssignmentEntry.getKey(); Map<String, List<Partition>> rsAssignment = rsAssignmentEntry.getValue(); if (!currReplicaSetIDs.contains(rsId)) { throw new IllegalStateException("current replica sets don't contain rs:" + rsId); } Map<String, List<Partition>> newRsTaskMap = newPlan.get(rsId); if (newRsTaskMap == null) { newRsTaskMap = Maps.newHashMap(); newPlan.put(rsId, newRsTaskMap); } for (Entry<String, List<Partition>> taskEntry : rsAssignment.entrySet()) { String cubeName = taskEntry.getKey(); List<Partition> partitions = taskEntry.getValue(); if (partitionChangeCubes.contains(cubeName)) { continue; } if (newRsTaskMap.size() < avgTasks) { newRsTaskMap.put(cubeName, partitions); } else { List<List<Partition>> cubeTasks = requireReassignTaskMap.get(cubeName); if (cubeTasks == null) { cubeTasks = Lists.newArrayList(); requireReassignTaskMap.put(cubeName, cubeTasks); } cubeTasks.add(partitions); } } } for (Entry<String, List<List<Partition>>> requireReassignTaskEntry : requireReassignTaskMap.entrySet()) { setNewPlanForCube(requireReassignTaskEntry.getKey(), requireReassignTaskEntry.getValue(), replicaSets, newPlan); } return newPlan; } }
DefaultAssigner implements Assigner { @Override public Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets, List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments) { Map<Integer, Map<String, List<Partition>>> newPlan = Maps.newHashMap(); if (replicaSets == null || cubes == null || cubes.size() == 0 || replicaSets.size() == 0) { return newPlan; } Map<Integer, Map<String, List<Partition>>> existingRSAssignmentsMap = AssignmentUtil .convertCubeAssign2ReplicaSetAssign(existingAssignments); Map<String, CubeAssignment> cubeAssignmentMap = Maps.newHashMap(); for (CubeAssignment existingAssignment : existingAssignments) { cubeAssignmentMap.put(existingAssignment.getCubeName(), existingAssignment); } Set<Integer> currReplicaSetIDs = Sets.newHashSet(); for (ReplicaSet replicaSet : replicaSets) { currReplicaSetIDs.add(replicaSet.getReplicaSetID()); } int totalTasks = 0; Map<String, List<List<Partition>>> requireReassignTaskMap = Maps.newHashMap(); Set<String> partitionChangeCubes = Sets.newHashSet(); for (StreamingCubeInfo cube : cubes) { CubeAssignment existingAssignment = cubeAssignmentMap.get(cube.getCubeName()); int prevPartitionNum = existingAssignment.getPartitionIDs().size(); int currPartitionNum = cube.getStreamingTableSourceInfo().getPartitions().size(); if (prevPartitionNum != currPartitionNum) { List<List<Partition>> cubeConsumeTasks = splitCubeConsumeTasks(cube, replicaSets.size()); requireReassignTaskMap.put(cube.getCubeName(), cubeConsumeTasks); partitionChangeCubes.add(cube.getCubeName()); } totalTasks += cube.getNumOfConsumerTasks(); } int avgTasks = totalTasks / replicaSets.size(); for (Entry<Integer, Map<String, List<Partition>>> rsAssignmentEntry : existingRSAssignmentsMap.entrySet()) { Integer rsId = rsAssignmentEntry.getKey(); Map<String, List<Partition>> rsAssignment = rsAssignmentEntry.getValue(); if (!currReplicaSetIDs.contains(rsId)) { throw new IllegalStateException("current replica sets don't contain rs:" + rsId); } Map<String, List<Partition>> newRsTaskMap = newPlan.get(rsId); if (newRsTaskMap == null) { newRsTaskMap = Maps.newHashMap(); newPlan.put(rsId, newRsTaskMap); } for (Entry<String, List<Partition>> taskEntry : rsAssignment.entrySet()) { String cubeName = taskEntry.getKey(); List<Partition> partitions = taskEntry.getValue(); if (partitionChangeCubes.contains(cubeName)) { continue; } if (newRsTaskMap.size() < avgTasks) { newRsTaskMap.put(cubeName, partitions); } else { List<List<Partition>> cubeTasks = requireReassignTaskMap.get(cubeName); if (cubeTasks == null) { cubeTasks = Lists.newArrayList(); requireReassignTaskMap.put(cubeName, cubeTasks); } cubeTasks.add(partitions); } } } for (Entry<String, List<List<Partition>>> requireReassignTaskEntry : requireReassignTaskMap.entrySet()) { setNewPlanForCube(requireReassignTaskEntry.getKey(), requireReassignTaskEntry.getValue(), replicaSets, newPlan); } return newPlan; }  }
DefaultAssigner implements Assigner { @Override public Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets, List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments) { Map<Integer, Map<String, List<Partition>>> newPlan = Maps.newHashMap(); if (replicaSets == null || cubes == null || cubes.size() == 0 || replicaSets.size() == 0) { return newPlan; } Map<Integer, Map<String, List<Partition>>> existingRSAssignmentsMap = AssignmentUtil .convertCubeAssign2ReplicaSetAssign(existingAssignments); Map<String, CubeAssignment> cubeAssignmentMap = Maps.newHashMap(); for (CubeAssignment existingAssignment : existingAssignments) { cubeAssignmentMap.put(existingAssignment.getCubeName(), existingAssignment); } Set<Integer> currReplicaSetIDs = Sets.newHashSet(); for (ReplicaSet replicaSet : replicaSets) { currReplicaSetIDs.add(replicaSet.getReplicaSetID()); } int totalTasks = 0; Map<String, List<List<Partition>>> requireReassignTaskMap = Maps.newHashMap(); Set<String> partitionChangeCubes = Sets.newHashSet(); for (StreamingCubeInfo cube : cubes) { CubeAssignment existingAssignment = cubeAssignmentMap.get(cube.getCubeName()); int prevPartitionNum = existingAssignment.getPartitionIDs().size(); int currPartitionNum = cube.getStreamingTableSourceInfo().getPartitions().size(); if (prevPartitionNum != currPartitionNum) { List<List<Partition>> cubeConsumeTasks = splitCubeConsumeTasks(cube, replicaSets.size()); requireReassignTaskMap.put(cube.getCubeName(), cubeConsumeTasks); partitionChangeCubes.add(cube.getCubeName()); } totalTasks += cube.getNumOfConsumerTasks(); } int avgTasks = totalTasks / replicaSets.size(); for (Entry<Integer, Map<String, List<Partition>>> rsAssignmentEntry : existingRSAssignmentsMap.entrySet()) { Integer rsId = rsAssignmentEntry.getKey(); Map<String, List<Partition>> rsAssignment = rsAssignmentEntry.getValue(); if (!currReplicaSetIDs.contains(rsId)) { throw new IllegalStateException("current replica sets don't contain rs:" + rsId); } Map<String, List<Partition>> newRsTaskMap = newPlan.get(rsId); if (newRsTaskMap == null) { newRsTaskMap = Maps.newHashMap(); newPlan.put(rsId, newRsTaskMap); } for (Entry<String, List<Partition>> taskEntry : rsAssignment.entrySet()) { String cubeName = taskEntry.getKey(); List<Partition> partitions = taskEntry.getValue(); if (partitionChangeCubes.contains(cubeName)) { continue; } if (newRsTaskMap.size() < avgTasks) { newRsTaskMap.put(cubeName, partitions); } else { List<List<Partition>> cubeTasks = requireReassignTaskMap.get(cubeName); if (cubeTasks == null) { cubeTasks = Lists.newArrayList(); requireReassignTaskMap.put(cubeName, cubeTasks); } cubeTasks.add(partitions); } } } for (Entry<String, List<List<Partition>>> requireReassignTaskEntry : requireReassignTaskMap.entrySet()) { setNewPlanForCube(requireReassignTaskEntry.getKey(), requireReassignTaskEntry.getValue(), replicaSets, newPlan); } return newPlan; }  @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 Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets, List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments) { Map<Integer, Map<String, List<Partition>>> newPlan = Maps.newHashMap(); if (replicaSets == null || cubes == null || cubes.size() == 0 || replicaSets.size() == 0) { return newPlan; } Map<Integer, Map<String, List<Partition>>> existingRSAssignmentsMap = AssignmentUtil .convertCubeAssign2ReplicaSetAssign(existingAssignments); Map<String, CubeAssignment> cubeAssignmentMap = Maps.newHashMap(); for (CubeAssignment existingAssignment : existingAssignments) { cubeAssignmentMap.put(existingAssignment.getCubeName(), existingAssignment); } Set<Integer> currReplicaSetIDs = Sets.newHashSet(); for (ReplicaSet replicaSet : replicaSets) { currReplicaSetIDs.add(replicaSet.getReplicaSetID()); } int totalTasks = 0; Map<String, List<List<Partition>>> requireReassignTaskMap = Maps.newHashMap(); Set<String> partitionChangeCubes = Sets.newHashSet(); for (StreamingCubeInfo cube : cubes) { CubeAssignment existingAssignment = cubeAssignmentMap.get(cube.getCubeName()); int prevPartitionNum = existingAssignment.getPartitionIDs().size(); int currPartitionNum = cube.getStreamingTableSourceInfo().getPartitions().size(); if (prevPartitionNum != currPartitionNum) { List<List<Partition>> cubeConsumeTasks = splitCubeConsumeTasks(cube, replicaSets.size()); requireReassignTaskMap.put(cube.getCubeName(), cubeConsumeTasks); partitionChangeCubes.add(cube.getCubeName()); } totalTasks += cube.getNumOfConsumerTasks(); } int avgTasks = totalTasks / replicaSets.size(); for (Entry<Integer, Map<String, List<Partition>>> rsAssignmentEntry : existingRSAssignmentsMap.entrySet()) { Integer rsId = rsAssignmentEntry.getKey(); Map<String, List<Partition>> rsAssignment = rsAssignmentEntry.getValue(); if (!currReplicaSetIDs.contains(rsId)) { throw new IllegalStateException("current replica sets don't contain rs:" + rsId); } Map<String, List<Partition>> newRsTaskMap = newPlan.get(rsId); if (newRsTaskMap == null) { newRsTaskMap = Maps.newHashMap(); newPlan.put(rsId, newRsTaskMap); } for (Entry<String, List<Partition>> taskEntry : rsAssignment.entrySet()) { String cubeName = taskEntry.getKey(); List<Partition> partitions = taskEntry.getValue(); if (partitionChangeCubes.contains(cubeName)) { continue; } if (newRsTaskMap.size() < avgTasks) { newRsTaskMap.put(cubeName, partitions); } else { List<List<Partition>> cubeTasks = requireReassignTaskMap.get(cubeName); if (cubeTasks == null) { cubeTasks = Lists.newArrayList(); requireReassignTaskMap.put(cubeName, cubeTasks); } cubeTasks.add(partitions); } } } for (Entry<String, List<List<Partition>>> requireReassignTaskEntry : requireReassignTaskMap.entrySet()) { setNewPlanForCube(requireReassignTaskEntry.getKey(), requireReassignTaskEntry.getValue(), replicaSets, newPlan); } return newPlan; }  @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);  }