@Test public void testCreateNewTasksOnlyForNewInstanceWhenInstanceIsAdded() { List<String> instances = Arrays.asList("instance1", "instance2", "instance3"); String instance4 = "instance4"; List<DatastreamGroup> datastreams = generateDatastreams("ds", 5); BroadcastStrategy strategy = new BroadcastStrategy(Optional.empty()); Map<String, Set<DatastreamTask>> assignment = strategy.assign(datastreams, instances, new HashMap<>()); List<String> newInstances = new ArrayList<>(instances); newInstances.add(instance4); Map<String, Set<DatastreamTask>> newAssignment = strategy.assign(datastreams, newInstances, assignment); for (String instance : instances) { Set<DatastreamTask> oldAssignmentTasks = assignment.get(instance); Set<DatastreamTask> newAssignmentTasks = newAssignment.get(instance); Assert.assertEquals(oldAssignmentTasks.size(), newAssignmentTasks.size()); Assert.assertTrue(newAssignmentTasks.containsAll(oldAssignmentTasks)); } Assert.assertEquals(newAssignment.get(instance4).size(), datastreams.size()); }
@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("Assigning {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Set<DatastreamTask> tasksAvailableToReuse = new HashSet<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> { currentAssignmentCopy.put(k, new HashSet<>(v)); tasksAvailableToReuse.addAll(v); }); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); Set<DatastreamTask> instanceTasks = currentAssignmentCopy.computeIfAbsent(instance, i -> new HashSet<>()); tasksAvailableToReuse.removeAll(instanceTasks); } Map<String, List<DatastreamTask>> reuseTaskMap = tasksAvailableToReuse.stream().collect(Collectors.groupingBy(DatastreamTask::getTaskPrefix)); int instancePos = 0; for (DatastreamGroup dg : datastreams) { List<DatastreamTask> reuseTasksPerDg = reuseTaskMap.getOrDefault(dg.getTaskPrefix(), Collections.emptyList()); int numTasks = getNumTasks(dg, instances.size()); for (int taskPos = 0; taskPos < numTasks; taskPos++) { String instance = instances.get(instancePos); DatastreamTask foundDatastreamTask = currentAssignmentCopy.get(instance) .stream() .filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix())) .findFirst() .orElseGet(() -> getOrCreateDatastreamTask(reuseTasksPerDg, dg)); currentAssignmentCopy.get(instance).remove(foundDatastreamTask); newAssignment.get(instance).add(foundDatastreamTask); instancePos = (instancePos + 1) % instances.size(); } } LOG.info("New assignment is {}", newAssignment); return newAssignment; }
BroadcastStrategy 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("Assigning {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Set<DatastreamTask> tasksAvailableToReuse = new HashSet<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> { currentAssignmentCopy.put(k, new HashSet<>(v)); tasksAvailableToReuse.addAll(v); }); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); Set<DatastreamTask> instanceTasks = currentAssignmentCopy.computeIfAbsent(instance, i -> new HashSet<>()); tasksAvailableToReuse.removeAll(instanceTasks); } Map<String, List<DatastreamTask>> reuseTaskMap = tasksAvailableToReuse.stream().collect(Collectors.groupingBy(DatastreamTask::getTaskPrefix)); int instancePos = 0; for (DatastreamGroup dg : datastreams) { List<DatastreamTask> reuseTasksPerDg = reuseTaskMap.getOrDefault(dg.getTaskPrefix(), Collections.emptyList()); int numTasks = getNumTasks(dg, instances.size()); for (int taskPos = 0; taskPos < numTasks; taskPos++) { String instance = instances.get(instancePos); DatastreamTask foundDatastreamTask = currentAssignmentCopy.get(instance) .stream() .filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix())) .findFirst() .orElseGet(() -> getOrCreateDatastreamTask(reuseTasksPerDg, dg)); currentAssignmentCopy.get(instance).remove(foundDatastreamTask); newAssignment.get(instance).add(foundDatastreamTask); instancePos = (instancePos + 1) % instances.size(); } } LOG.info("New assignment is {}", newAssignment); return newAssignment; } }
BroadcastStrategy 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("Assigning {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Set<DatastreamTask> tasksAvailableToReuse = new HashSet<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> { currentAssignmentCopy.put(k, new HashSet<>(v)); tasksAvailableToReuse.addAll(v); }); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); Set<DatastreamTask> instanceTasks = currentAssignmentCopy.computeIfAbsent(instance, i -> new HashSet<>()); tasksAvailableToReuse.removeAll(instanceTasks); } Map<String, List<DatastreamTask>> reuseTaskMap = tasksAvailableToReuse.stream().collect(Collectors.groupingBy(DatastreamTask::getTaskPrefix)); int instancePos = 0; for (DatastreamGroup dg : datastreams) { List<DatastreamTask> reuseTasksPerDg = reuseTaskMap.getOrDefault(dg.getTaskPrefix(), Collections.emptyList()); int numTasks = getNumTasks(dg, instances.size()); for (int taskPos = 0; taskPos < numTasks; taskPos++) { String instance = instances.get(instancePos); DatastreamTask foundDatastreamTask = currentAssignmentCopy.get(instance) .stream() .filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix())) .findFirst() .orElseGet(() -> getOrCreateDatastreamTask(reuseTasksPerDg, dg)); currentAssignmentCopy.get(instance).remove(foundDatastreamTask); newAssignment.get(instance).add(foundDatastreamTask); instancePos = (instancePos + 1) % instances.size(); } } LOG.info("New assignment is {}", newAssignment); return newAssignment; } BroadcastStrategy(Optional<Integer> maxTasks); }
BroadcastStrategy 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("Assigning {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Set<DatastreamTask> tasksAvailableToReuse = new HashSet<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> { currentAssignmentCopy.put(k, new HashSet<>(v)); tasksAvailableToReuse.addAll(v); }); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); Set<DatastreamTask> instanceTasks = currentAssignmentCopy.computeIfAbsent(instance, i -> new HashSet<>()); tasksAvailableToReuse.removeAll(instanceTasks); } Map<String, List<DatastreamTask>> reuseTaskMap = tasksAvailableToReuse.stream().collect(Collectors.groupingBy(DatastreamTask::getTaskPrefix)); int instancePos = 0; for (DatastreamGroup dg : datastreams) { List<DatastreamTask> reuseTasksPerDg = reuseTaskMap.getOrDefault(dg.getTaskPrefix(), Collections.emptyList()); int numTasks = getNumTasks(dg, instances.size()); for (int taskPos = 0; taskPos < numTasks; taskPos++) { String instance = instances.get(instancePos); DatastreamTask foundDatastreamTask = currentAssignmentCopy.get(instance) .stream() .filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix())) .findFirst() .orElseGet(() -> getOrCreateDatastreamTask(reuseTasksPerDg, dg)); currentAssignmentCopy.get(instance).remove(foundDatastreamTask); newAssignment.get(instance).add(foundDatastreamTask); instancePos = (instancePos + 1) % instances.size(); } } LOG.info("New assignment is {}", newAssignment); return newAssignment; } BroadcastStrategy(Optional<Integer> maxTasks); @Override Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances,
      Map<String, Set<DatastreamTask>> currentAssignment); }
BroadcastStrategy 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("Assigning {} datastreams to {} instances with {} tasks", datastreams.size(), instances.size(), totalAssignedTasks); Map<String, Set<DatastreamTask>> newAssignment = new HashMap<>(); Set<DatastreamTask> tasksAvailableToReuse = new HashSet<>(); Map<String, Set<DatastreamTask>> currentAssignmentCopy = new HashMap<>(currentAssignment.size()); currentAssignment.forEach((k, v) -> { currentAssignmentCopy.put(k, new HashSet<>(v)); tasksAvailableToReuse.addAll(v); }); for (String instance : instances) { newAssignment.put(instance, new HashSet<>()); Set<DatastreamTask> instanceTasks = currentAssignmentCopy.computeIfAbsent(instance, i -> new HashSet<>()); tasksAvailableToReuse.removeAll(instanceTasks); } Map<String, List<DatastreamTask>> reuseTaskMap = tasksAvailableToReuse.stream().collect(Collectors.groupingBy(DatastreamTask::getTaskPrefix)); int instancePos = 0; for (DatastreamGroup dg : datastreams) { List<DatastreamTask> reuseTasksPerDg = reuseTaskMap.getOrDefault(dg.getTaskPrefix(), Collections.emptyList()); int numTasks = getNumTasks(dg, instances.size()); for (int taskPos = 0; taskPos < numTasks; taskPos++) { String instance = instances.get(instancePos); DatastreamTask foundDatastreamTask = currentAssignmentCopy.get(instance) .stream() .filter(x -> x.getTaskPrefix().equals(dg.getTaskPrefix())) .findFirst() .orElseGet(() -> getOrCreateDatastreamTask(reuseTasksPerDg, dg)); currentAssignmentCopy.get(instance).remove(foundDatastreamTask); newAssignment.get(instance).add(foundDatastreamTask); instancePos = (instancePos + 1) % instances.size(); } } LOG.info("New assignment is {}", newAssignment); return newAssignment; } BroadcastStrategy(Optional<Integer> maxTasks); @Override Map<String, Set<DatastreamTask>> assign(List<DatastreamGroup> datastreams, List<String> instances,
      Map<String, Set<DatastreamTask>> currentAssignment);  }