@Test public void testDontCreateNewTasksWhenInstanceChanged() { String[] instances = new String[]{"instance1", "instance2", "instance3"}; List<DatastreamGroup> datastreams = generateDatastreams("ds", 5); StickyMulticastStrategy strategy = new StickyMulticastStrategy(Optional.empty(), Optional.empty()); Map<String, Set<DatastreamTask>> assignment = strategy.assign(datastreams, Arrays.asList(instances), new HashMap<>()); instances = new String[]{"instance1", "instance2"}; Map<String, Set<DatastreamTask>> newAssignment = strategy.assign(datastreams, Arrays.asList(instances), assignment); Set<DatastreamTask> oldAssignmentTasks = assignment.values().stream().flatMap(Set::stream).collect(Collectors.toSet()); Set<DatastreamTask> newAssignmentTasks = newAssignment.values().stream().flatMap(Set::stream).collect(Collectors.toSet()); Assert.assertTrue(oldAssignmentTasks.containsAll(newAssignmentTasks)); }
@Override public Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances, Map<String, Set<DatastreamTask>> currentAssignment) { int totalAssignedTasks = currentAssignment.values().stream().mapToInt(Set::size).sum(); LOG.info("Begin assign {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); if (instances.isEmpty()) { return Collections.emptyMap(); } Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> currentAssignmentCopy.put(k, new HashSet<>(v))); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); } Map<DatastreamGroup, Integer> unallocated = new HashMap<>(); Map<DatastreamGroup, List<DatastreamTask>> tasksNeedToRelocate = new HashMap<>(); for (DatastreamGroup dg : datastreams) { int numTasks = getNumTasks(dg, instances.size()); Set<DatastreamTask> allAliveTasks = new HashSet<>(); for (String instance : instances) { if (numTasks <= 0) { break; } List<DatastreamTask> foundDatastreamTasks = Optional.ofNullable(currentAssignmentCopy.get(instance)).map(c -> c.stream().filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix()) && !allAliveTasks.contains(x)) .collect(Collectors.toList())).orElse(Collections.emptyList()); allAliveTasks.addAll(foundDatastreamTasks); if (!foundDatastreamTasks.isEmpty()) { newAssignment.get(instance).addAll(foundDatastreamTasks); currentAssignmentCopy.get(instance).removeAll(foundDatastreamTasks); numTasks -= foundDatastreamTasks.size(); } } Set<DatastreamTask> unallocatedTasks = currentAssignmentCopy.values().stream().flatMap(Set::stream) .filter(dg::belongsTo).collect(Collectors.toSet()); unallocatedTasks.removeAll(allAliveTasks); tasksNeedToRelocate.put(dg, new ArrayList<>(unallocatedTasks)); if (numTasks > 0) { unallocated.put(dg, numTasks); } } List<String> instancesBySize = new ArrayList<>(instances); Collections.shuffle(instancesBySize); instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); for (Map.Entry<DatastreamGroup, Integer> entry : unallocated.entrySet()) { DatastreamGroup dg = entry.getKey(); int pendingTasks = entry.getValue(); List<DatastreamTask> unallocatedTasks = tasksNeedToRelocate.get(dg); while (pendingTasks > 0) { for (String instance : instancesBySize) { DatastreamTask task = unallocatedTasks.size() > 0 ? unallocatedTasks.remove(unallocatedTasks.size() - 1) : new DatastreamTaskImpl(dg.getDatastreams()); newAssignment.get(instance).add(task); pendingTasks--; if (pendingTasks == 0) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } if (newAssignment.get(instancesBySize.get(instancesBySize.size() - 1)).size() - newAssignment.get(instancesBySize.get(0)).size() > _imbalanceThreshold) { int tasksTotal = newAssignment.values().stream().mapToInt(Set::size).sum(); int minTasksPerInstance = tasksTotal / instances.size(); while (newAssignment.get(instancesBySize.get(0)).size() < minTasksPerInstance) { String smallInstance = instancesBySize.get(0); String largeInstance = instancesBySize.get(instancesBySize.size() - 1); for (DatastreamTask task : new ArrayList<>(newAssignment.get(largeInstance))) { newAssignment.get(largeInstance).remove(task); newAssignment.get(smallInstance).add(task); if (newAssignment.get(smallInstance).size() >= minTasksPerInstance || newAssignment.get(largeInstance).size() <= minTasksPerInstance + 1) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } LOG.info("Assignment completed"); LOG.debug("New assignment is {}", newAssignment); sanityChecks(datastreams, instances, newAssignment); return newAssignment; }
StickyMulticastStrategy implements AssignmentStrategy { @Override public Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances, Map<String, Set<DatastreamTask>> currentAssignment) { int totalAssignedTasks = currentAssignment.values().stream().mapToInt(Set::size).sum(); LOG.info("Begin assign {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); if (instances.isEmpty()) { return Collections.emptyMap(); } Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> currentAssignmentCopy.put(k, new HashSet<>(v))); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); } Map<DatastreamGroup, Integer> unallocated = new HashMap<>(); Map<DatastreamGroup, List<DatastreamTask>> tasksNeedToRelocate = new HashMap<>(); for (DatastreamGroup dg : datastreams) { int numTasks = getNumTasks(dg, instances.size()); Set<DatastreamTask> allAliveTasks = new HashSet<>(); for (String instance : instances) { if (numTasks <= 0) { break; } List<DatastreamTask> foundDatastreamTasks = Optional.ofNullable(currentAssignmentCopy.get(instance)).map(c -> c.stream().filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix()) && !allAliveTasks.contains(x)) .collect(Collectors.toList())).orElse(Collections.emptyList()); allAliveTasks.addAll(foundDatastreamTasks); if (!foundDatastreamTasks.isEmpty()) { newAssignment.get(instance).addAll(foundDatastreamTasks); currentAssignmentCopy.get(instance).removeAll(foundDatastreamTasks); numTasks -= foundDatastreamTasks.size(); } } Set<DatastreamTask> unallocatedTasks = currentAssignmentCopy.values().stream().flatMap(Set::stream) .filter(dg::belongsTo).collect(Collectors.toSet()); unallocatedTasks.removeAll(allAliveTasks); tasksNeedToRelocate.put(dg, new ArrayList<>(unallocatedTasks)); if (numTasks > 0) { unallocated.put(dg, numTasks); } } List<String> instancesBySize = new ArrayList<>(instances); Collections.shuffle(instancesBySize); instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); for (Map.Entry<DatastreamGroup, Integer> entry : unallocated.entrySet()) { DatastreamGroup dg = entry.getKey(); int pendingTasks = entry.getValue(); List<DatastreamTask> unallocatedTasks = tasksNeedToRelocate.get(dg); while (pendingTasks > 0) { for (String instance : instancesBySize) { DatastreamTask task = unallocatedTasks.size() > 0 ? unallocatedTasks.remove(unallocatedTasks.size() - 1) : new DatastreamTaskImpl(dg.getDatastreams()); newAssignment.get(instance).add(task); pendingTasks--; if (pendingTasks == 0) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } if (newAssignment.get(instancesBySize.get(instancesBySize.size() - 1)).size() - newAssignment.get(instancesBySize.get(0)).size() > _imbalanceThreshold) { int tasksTotal = newAssignment.values().stream().mapToInt(Set::size).sum(); int minTasksPerInstance = tasksTotal / instances.size(); while (newAssignment.get(instancesBySize.get(0)).size() < minTasksPerInstance) { String smallInstance = instancesBySize.get(0); String largeInstance = instancesBySize.get(instancesBySize.size() - 1); for (DatastreamTask task : new ArrayList<>(newAssignment.get(largeInstance))) { newAssignment.get(largeInstance).remove(task); newAssignment.get(smallInstance).add(task); if (newAssignment.get(smallInstance).size() >= minTasksPerInstance || newAssignment.get(largeInstance).size() <= minTasksPerInstance + 1) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } LOG.info("Assignment completed"); LOG.debug("New assignment is {}", newAssignment); sanityChecks(datastreams, instances, newAssignment); return newAssignment; } }
StickyMulticastStrategy implements AssignmentStrategy { @Override public Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances, Map<String, Set<DatastreamTask>> currentAssignment) { int totalAssignedTasks = currentAssignment.values().stream().mapToInt(Set::size).sum(); LOG.info("Begin assign {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); if (instances.isEmpty()) { return Collections.emptyMap(); } Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> currentAssignmentCopy.put(k, new HashSet<>(v))); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); } Map<DatastreamGroup, Integer> unallocated = new HashMap<>(); Map<DatastreamGroup, List<DatastreamTask>> tasksNeedToRelocate = new HashMap<>(); for (DatastreamGroup dg : datastreams) { int numTasks = getNumTasks(dg, instances.size()); Set<DatastreamTask> allAliveTasks = new HashSet<>(); for (String instance : instances) { if (numTasks <= 0) { break; } List<DatastreamTask> foundDatastreamTasks = Optional.ofNullable(currentAssignmentCopy.get(instance)).map(c -> c.stream().filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix()) && !allAliveTasks.contains(x)) .collect(Collectors.toList())).orElse(Collections.emptyList()); allAliveTasks.addAll(foundDatastreamTasks); if (!foundDatastreamTasks.isEmpty()) { newAssignment.get(instance).addAll(foundDatastreamTasks); currentAssignmentCopy.get(instance).removeAll(foundDatastreamTasks); numTasks -= foundDatastreamTasks.size(); } } Set<DatastreamTask> unallocatedTasks = currentAssignmentCopy.values().stream().flatMap(Set::stream) .filter(dg::belongsTo).collect(Collectors.toSet()); unallocatedTasks.removeAll(allAliveTasks); tasksNeedToRelocate.put(dg, new ArrayList<>(unallocatedTasks)); if (numTasks > 0) { unallocated.put(dg, numTasks); } } List<String> instancesBySize = new ArrayList<>(instances); Collections.shuffle(instancesBySize); instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); for (Map.Entry<DatastreamGroup, Integer> entry : unallocated.entrySet()) { DatastreamGroup dg = entry.getKey(); int pendingTasks = entry.getValue(); List<DatastreamTask> unallocatedTasks = tasksNeedToRelocate.get(dg); while (pendingTasks > 0) { for (String instance : instancesBySize) { DatastreamTask task = unallocatedTasks.size() > 0 ? unallocatedTasks.remove(unallocatedTasks.size() - 1) : new DatastreamTaskImpl(dg.getDatastreams()); newAssignment.get(instance).add(task); pendingTasks--; if (pendingTasks == 0) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } if (newAssignment.get(instancesBySize.get(instancesBySize.size() - 1)).size() - newAssignment.get(instancesBySize.get(0)).size() > _imbalanceThreshold) { int tasksTotal = newAssignment.values().stream().mapToInt(Set::size).sum(); int minTasksPerInstance = tasksTotal / instances.size(); while (newAssignment.get(instancesBySize.get(0)).size() < minTasksPerInstance) { String smallInstance = instancesBySize.get(0); String largeInstance = instancesBySize.get(instancesBySize.size() - 1); for (DatastreamTask task : new ArrayList<>(newAssignment.get(largeInstance))) { newAssignment.get(largeInstance).remove(task); newAssignment.get(smallInstance).add(task); if (newAssignment.get(smallInstance).size() >= minTasksPerInstance || newAssignment.get(largeInstance).size() <= minTasksPerInstance + 1) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } LOG.info("Assignment completed"); LOG.debug("New assignment is {}", newAssignment); sanityChecks(datastreams, instances, newAssignment); return newAssignment; } StickyMulticastStrategy(Optional<Integer> maxTasks, Optional<Integer> imbalanceThreshold); }
StickyMulticastStrategy implements AssignmentStrategy { @Override public Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances, Map<String, Set<DatastreamTask>> currentAssignment) { int totalAssignedTasks = currentAssignment.values().stream().mapToInt(Set::size).sum(); LOG.info("Begin assign {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); if (instances.isEmpty()) { return Collections.emptyMap(); } Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> currentAssignmentCopy.put(k, new HashSet<>(v))); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); } Map<DatastreamGroup, Integer> unallocated = new HashMap<>(); Map<DatastreamGroup, List<DatastreamTask>> tasksNeedToRelocate = new HashMap<>(); for (DatastreamGroup dg : datastreams) { int numTasks = getNumTasks(dg, instances.size()); Set<DatastreamTask> allAliveTasks = new HashSet<>(); for (String instance : instances) { if (numTasks <= 0) { break; } List<DatastreamTask> foundDatastreamTasks = Optional.ofNullable(currentAssignmentCopy.get(instance)).map(c -> c.stream().filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix()) && !allAliveTasks.contains(x)) .collect(Collectors.toList())).orElse(Collections.emptyList()); allAliveTasks.addAll(foundDatastreamTasks); if (!foundDatastreamTasks.isEmpty()) { newAssignment.get(instance).addAll(foundDatastreamTasks); currentAssignmentCopy.get(instance).removeAll(foundDatastreamTasks); numTasks -= foundDatastreamTasks.size(); } } Set<DatastreamTask> unallocatedTasks = currentAssignmentCopy.values().stream().flatMap(Set::stream) .filter(dg::belongsTo).collect(Collectors.toSet()); unallocatedTasks.removeAll(allAliveTasks); tasksNeedToRelocate.put(dg, new ArrayList<>(unallocatedTasks)); if (numTasks > 0) { unallocated.put(dg, numTasks); } } List<String> instancesBySize = new ArrayList<>(instances); Collections.shuffle(instancesBySize); instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); for (Map.Entry<DatastreamGroup, Integer> entry : unallocated.entrySet()) { DatastreamGroup dg = entry.getKey(); int pendingTasks = entry.getValue(); List<DatastreamTask> unallocatedTasks = tasksNeedToRelocate.get(dg); while (pendingTasks > 0) { for (String instance : instancesBySize) { DatastreamTask task = unallocatedTasks.size() > 0 ? unallocatedTasks.remove(unallocatedTasks.size() - 1) : new DatastreamTaskImpl(dg.getDatastreams()); newAssignment.get(instance).add(task); pendingTasks--; if (pendingTasks == 0) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } if (newAssignment.get(instancesBySize.get(instancesBySize.size() - 1)).size() - newAssignment.get(instancesBySize.get(0)).size() > _imbalanceThreshold) { int tasksTotal = newAssignment.values().stream().mapToInt(Set::size).sum(); int minTasksPerInstance = tasksTotal / instances.size(); while (newAssignment.get(instancesBySize.get(0)).size() < minTasksPerInstance) { String smallInstance = instancesBySize.get(0); String largeInstance = instancesBySize.get(instancesBySize.size() - 1); for (DatastreamTask task : new ArrayList<>(newAssignment.get(largeInstance))) { newAssignment.get(largeInstance).remove(task); newAssignment.get(smallInstance).add(task); if (newAssignment.get(smallInstance).size() >= minTasksPerInstance || newAssignment.get(largeInstance).size() <= minTasksPerInstance + 1) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } LOG.info("Assignment completed"); LOG.debug("New assignment is {}", newAssignment); sanityChecks(datastreams, instances, newAssignment); return newAssignment; } StickyMulticastStrategy(Optional<Integer> maxTasks, Optional<Integer> imbalanceThreshold); @Override Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances,
      Map<String, Set<DatastreamTask>> currentAssignment); }
StickyMulticastStrategy implements AssignmentStrategy { @Override public Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances, Map<String, Set<DatastreamTask>> currentAssignment) { int totalAssignedTasks = currentAssignment.values().stream().mapToInt(Set::size).sum(); LOG.info("Begin assign {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); if (instances.isEmpty()) { return Collections.emptyMap(); } Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> currentAssignmentCopy.put(k, new HashSet<>(v))); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); } Map<DatastreamGroup, Integer> unallocated = new HashMap<>(); Map<DatastreamGroup, List<DatastreamTask>> tasksNeedToRelocate = new HashMap<>(); for (DatastreamGroup dg : datastreams) { int numTasks = getNumTasks(dg, instances.size()); Set<DatastreamTask> allAliveTasks = new HashSet<>(); for (String instance : instances) { if (numTasks <= 0) { break; } List<DatastreamTask> foundDatastreamTasks = Optional.ofNullable(currentAssignmentCopy.get(instance)).map(c -> c.stream().filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix()) && !allAliveTasks.contains(x)) .collect(Collectors.toList())).orElse(Collections.emptyList()); allAliveTasks.addAll(foundDatastreamTasks); if (!foundDatastreamTasks.isEmpty()) { newAssignment.get(instance).addAll(foundDatastreamTasks); currentAssignmentCopy.get(instance).removeAll(foundDatastreamTasks); numTasks -= foundDatastreamTasks.size(); } } Set<DatastreamTask> unallocatedTasks = currentAssignmentCopy.values().stream().flatMap(Set::stream) .filter(dg::belongsTo).collect(Collectors.toSet()); unallocatedTasks.removeAll(allAliveTasks); tasksNeedToRelocate.put(dg, new ArrayList<>(unallocatedTasks)); if (numTasks > 0) { unallocated.put(dg, numTasks); } } List<String> instancesBySize = new ArrayList<>(instances); Collections.shuffle(instancesBySize); instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); for (Map.Entry<DatastreamGroup, Integer> entry : unallocated.entrySet()) { DatastreamGroup dg = entry.getKey(); int pendingTasks = entry.getValue(); List<DatastreamTask> unallocatedTasks = tasksNeedToRelocate.get(dg); while (pendingTasks > 0) { for (String instance : instancesBySize) { DatastreamTask task = unallocatedTasks.size() > 0 ? unallocatedTasks.remove(unallocatedTasks.size() - 1) : new DatastreamTaskImpl(dg.getDatastreams()); newAssignment.get(instance).add(task); pendingTasks--; if (pendingTasks == 0) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } if (newAssignment.get(instancesBySize.get(instancesBySize.size() - 1)).size() - newAssignment.get(instancesBySize.get(0)).size() > _imbalanceThreshold) { int tasksTotal = newAssignment.values().stream().mapToInt(Set::size).sum(); int minTasksPerInstance = tasksTotal / instances.size(); while (newAssignment.get(instancesBySize.get(0)).size() < minTasksPerInstance) { String smallInstance = instancesBySize.get(0); String largeInstance = instancesBySize.get(instancesBySize.size() - 1); for (DatastreamTask task : new ArrayList<>(newAssignment.get(largeInstance))) { newAssignment.get(largeInstance).remove(task); newAssignment.get(smallInstance).add(task); if (newAssignment.get(smallInstance).size() >= minTasksPerInstance || newAssignment.get(largeInstance).size() <= minTasksPerInstance + 1) { break; } } instancesBySize.sort(Comparator.comparing(x -> newAssignment.get(x).size())); } } LOG.info("Assignment completed"); LOG.debug("New assignment is {}", newAssignment); sanityChecks(datastreams, instances, newAssignment); return newAssignment; } StickyMulticastStrategy(Optional<Integer> maxTasks, Optional<Integer> imbalanceThreshold); @Override Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances,
      Map<String, Set<DatastreamTask>> currentAssignment);  }