@Test public void testParitionAndConsumerAddAndDelete() { LeastAdjustmentConsumerPartitionAssigningStrategy strategy = new LeastAdjustmentConsumerPartitionAssigningStrategy(); AssignBalancer assignBalancer = new LeastAdjustmentAssianBalancer(); Reflects.forField().setDeclaredFieldValue(LeastAdjustmentConsumerPartitionAssigningStrategy.class, "m_assignBalancer", strategy, assignBalancer); List<Partition> partitions = new ArrayList<>(); for (int i = 0; i < 3; i++) { Partition p = new Partition(i); partitions.add(p); } partitions.add(new Partition(5)); partitions.add(new Partition(10)); partitions.add(new Partition(11)); partitions.add(new Partition(12)); Map<String, ClientContext> currentConsumers = new HashMap<>(); final ClientContext a = new ClientContext("a", "a", 1, "a", "a", 1L); final ClientContext b = new ClientContext("b", "a", 1, "a", "a", 1L); final ClientContext c = new ClientContext("c", "a", 1, "a", "a", 1L); final ClientContext d = new ClientContext("d", "a", 1, "a", "a", 1L); final ClientContext e = new ClientContext("e", "a", 1, "a", "a", 1L); final ClientContext f = new ClientContext("f", "a", 1, "a", "a", 1L); currentConsumers.put("a", a); currentConsumers.put("b", b); currentConsumers.put("c", c); currentConsumers.put("f", f); Map<Integer, Map<String, ClientContext>> originAssigns = new HashMap<>(); originAssigns.put(0, new HashMap<String, ClientContext>() { { put("a", a); } }); originAssigns.put(1, new HashMap<String, ClientContext>() { { put("b", b); } }); originAssigns.put(2, new HashMap<String, ClientContext>() { { put("c", c); } }); originAssigns.put(3, new HashMap<String, ClientContext>() { { put("d", d); } }); originAssigns.put(4, new HashMap<String, ClientContext>() { { put("e", e); } }); originAssigns.put(5, new HashMap<String, ClientContext>() { { put("a", a); } }); Map<Integer, Map<String, ClientContext>> newAssigns = strategy .assign(partitions, currentConsumers, originAssigns); System.out.println("TestConsumerAddAndDelete:"); System.out.println("---------------------------------------------------------------"); System.out.println("Current Consumers:"); for (Entry<String, ClientContext> consumer : currentConsumers.entrySet()) { System.out.print(consumer.getKey() + " "); } System.out.println(); System.out.println("---------------------------------------------------------------"); System.out.println("Current Partitions:"); for (Partition partition : partitions) { System.out.print(partition.getId() + " "); } System.out.println(); System.out.println("---------------------------------------------------------------"); System.out.println("Origin Assignments:"); for (Entry<Integer, Map<String, ClientContext>> assign : originAssigns.entrySet()) { System.out.println(assign); } System.out.println("---------------------------------------------------------------"); System.out.println("New Assignments:"); for (Entry<Integer, Map<String, ClientContext>> assign : newAssigns.entrySet()) { System.out.println(assign); } normalAssert(partitions, currentConsumers, newAssigns); }
@Override public Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions, Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns) { Map<Integer, Map<String, ClientContext>> result = new HashMap<>(); if (partitions == null || partitions.isEmpty() || currentConsumers == null || currentConsumers.isEmpty()) { return result; } if (originAssigns == null) { originAssigns = Collections.emptyMap(); } Map<String, List<Integer>> originConsumerToPartition = mapConsumerToPartitions(partitions, currentConsumers, originAssigns); List<Integer> freePartitions = getFreePartitions(originConsumerToPartition, partitions); Map<String, List<Integer>> newAssigns = m_assignBalancer.assign(originConsumerToPartition, freePartitions); for (Entry<String, List<Integer>> assign : newAssigns.entrySet()) { putAssignToResult(result, currentConsumers, assign.getKey(), assign.getValue()); } return result; }
LeastAdjustmentConsumerPartitionAssigningStrategy implements ConsumerPartitionAssigningStrategy { @Override public Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions, Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns) { Map<Integer, Map<String, ClientContext>> result = new HashMap<>(); if (partitions == null || partitions.isEmpty() || currentConsumers == null || currentConsumers.isEmpty()) { return result; } if (originAssigns == null) { originAssigns = Collections.emptyMap(); } Map<String, List<Integer>> originConsumerToPartition = mapConsumerToPartitions(partitions, currentConsumers, originAssigns); List<Integer> freePartitions = getFreePartitions(originConsumerToPartition, partitions); Map<String, List<Integer>> newAssigns = m_assignBalancer.assign(originConsumerToPartition, freePartitions); for (Entry<String, List<Integer>> assign : newAssigns.entrySet()) { putAssignToResult(result, currentConsumers, assign.getKey(), assign.getValue()); } return result; } }
LeastAdjustmentConsumerPartitionAssigningStrategy implements ConsumerPartitionAssigningStrategy { @Override public Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions, Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns) { Map<Integer, Map<String, ClientContext>> result = new HashMap<>(); if (partitions == null || partitions.isEmpty() || currentConsumers == null || currentConsumers.isEmpty()) { return result; } if (originAssigns == null) { originAssigns = Collections.emptyMap(); } Map<String, List<Integer>> originConsumerToPartition = mapConsumerToPartitions(partitions, currentConsumers, originAssigns); List<Integer> freePartitions = getFreePartitions(originConsumerToPartition, partitions); Map<String, List<Integer>> newAssigns = m_assignBalancer.assign(originConsumerToPartition, freePartitions); for (Entry<String, List<Integer>> assign : newAssigns.entrySet()) { putAssignToResult(result, currentConsumers, assign.getKey(), assign.getValue()); } return result; }  }
LeastAdjustmentConsumerPartitionAssigningStrategy implements ConsumerPartitionAssigningStrategy { @Override public Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions, Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns) { Map<Integer, Map<String, ClientContext>> result = new HashMap<>(); if (partitions == null || partitions.isEmpty() || currentConsumers == null || currentConsumers.isEmpty()) { return result; } if (originAssigns == null) { originAssigns = Collections.emptyMap(); } Map<String, List<Integer>> originConsumerToPartition = mapConsumerToPartitions(partitions, currentConsumers, originAssigns); List<Integer> freePartitions = getFreePartitions(originConsumerToPartition, partitions); Map<String, List<Integer>> newAssigns = m_assignBalancer.assign(originConsumerToPartition, freePartitions); for (Entry<String, List<Integer>> assign : newAssigns.entrySet()) { putAssignToResult(result, currentConsumers, assign.getKey(), assign.getValue()); } return result; }  @Override Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions,
	      Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns); }
LeastAdjustmentConsumerPartitionAssigningStrategy implements ConsumerPartitionAssigningStrategy { @Override public Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions, Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns) { Map<Integer, Map<String, ClientContext>> result = new HashMap<>(); if (partitions == null || partitions.isEmpty() || currentConsumers == null || currentConsumers.isEmpty()) { return result; } if (originAssigns == null) { originAssigns = Collections.emptyMap(); } Map<String, List<Integer>> originConsumerToPartition = mapConsumerToPartitions(partitions, currentConsumers, originAssigns); List<Integer> freePartitions = getFreePartitions(originConsumerToPartition, partitions); Map<String, List<Integer>> newAssigns = m_assignBalancer.assign(originConsumerToPartition, freePartitions); for (Entry<String, List<Integer>> assign : newAssigns.entrySet()) { putAssignToResult(result, currentConsumers, assign.getKey(), assign.getValue()); } return result; }  @Override Map<Integer, Map<String, ClientContext>> assign(List<Partition> partitions,
	      Map<String, ClientContext> currentConsumers, Map<Integer, Map<String, ClientContext>> originAssigns);  }