@Test public void AddAndDeleteTPTest() { LeastAdjustmentBrokerPartitionAssigningStrategy strategy = new LeastAdjustmentBrokerPartitionAssigningStrategy(); AssignBalancer assignBalancer = new LeastAdjustmentAssianBalancer(); Reflects.forField().setDeclaredFieldValue(LeastAdjustmentBrokerPartitionAssigningStrategy.class, "m_assignBalancer", strategy, assignBalancer); Map<String, ClientContext> brokers = generateBrokers(5, "b"); Map<String, Assignment<Integer>> originAssignments = new HashMap<>(); Map<String, Topic> topicMap = generateTopicsWithIncrementalPartitionCount(10, "t"); List<Topic> topics = new ArrayList<>(topicMap.values()); int brokerPos = 0; int brokerListSize = brokers.size(); for (int i = 1; i <= 10; i++) { originAssignments.put("t" + i, new Assignment<Integer>()); Assignment<Integer> assignment = originAssignments.get("t" + i); for (int j = 0; j < i; j++) { Map<String, ClientContext> brokerAssign = new HashMap<>(); brokerAssign.put("b" + (brokerPos + 1), brokers.get("b" + (brokerPos + 1))); brokerPos = (brokerPos + 1) % brokerListSize; assignment.addAssignment(j, brokerAssign); } } topics.remove(new Topic("t1")); Topic t = new Topic("t11"); for (int i = 0; i < 10; i++) { t.addPartition(new Partition(i)); } topics.add(t); t = new Topic("t12"); for (int i = 0; i < 10; i++) { t.addPartition(new Partition(i)); } topics.add(t); Map<String, Assignment<Integer>> newAssignments = strategy.assign(brokers, topics, originAssignments); System.out.println("AddAndDeleteBrokerTest:"); System.out.println("---------------------------------------------------------------"); System.out.println("Current Topics:"); for (Topic topic : topics) { System.out.print(topic.getName() + " "); } System.out.println(); printResult(originAssignments, newAssignments); normalAssert(brokers, topics, newAssignments); assertFalse(newAssignments.keySet().contains("t1")); assertTrue(newAssignments.keySet().contains("t11")); assertTrue(newAssignments.keySet().contains("t12")); }
@Override public Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics, Map<String, Assignment<Integer>> originAssignments) { Map<String, Assignment<Integer>> newAssignments = new HashMap<String, Assignment<Integer>>(); if (brokers == null || brokers.isEmpty() || topics == null || topics.isEmpty()) { return newAssignments; } Map<String, List<Pair<String, Integer>>> avalibleOriginAssignments = new HashMap<String, List<Pair<String, Integer>>>(); List<Pair<String, Integer>> freeTps = new ArrayList<Pair<String, Integer>>(); for (String broker : brokers.keySet()) { avalibleOriginAssignments.put(broker, new ArrayList<Pair<String, Integer>>()); } for (Topic topic : topics) { if (originAssignments.containsKey(topic.getName())) { Assignment<Integer> topicOriginAssignments = originAssignments.get(topic.getName()); for (Partition partition : topic.getPartitions()) { Map<String, ClientContext> originTPAssianment = topicOriginAssignments.getAssignment(partition.getId()); if (originTPAssianment != null) { if (originTPAssianment.size() != 1) { log.error("TP have more than one broker assigned"); } ClientContext broker = originTPAssianment.values().iterator().next(); if (brokers.containsKey(broker.getName())) { avalibleOriginAssignments.get(broker.getName()).add( new Pair<String, Integer>(topic.getName(), partition.getId())); continue; } } freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } else { for (Partition partition : topic.getPartitions()) { freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } } Map<String, List<Pair<String, Integer>>> newAssigns = m_assignBalancer.assign(avalibleOriginAssignments, freeTps); for (Entry<String, List<Pair<String, Integer>>> assign : newAssigns.entrySet()) { for (Pair<String, Integer> tp : assign.getValue()) { if (!newAssignments.containsKey(tp.getKey())) { newAssignments.put(tp.getKey(), new Assignment<Integer>()); } Map<String, ClientContext> broker = new HashMap<String, ClientContext>(); broker.put(assign.getKey(), brokers.get(assign.getKey())); newAssignments.get(tp.getKey()).addAssignment(tp.getValue(), broker); } } return newAssignments; }
LeastAdjustmentBrokerPartitionAssigningStrategy implements BrokerPartitionAssigningStrategy { @Override public Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics, Map<String, Assignment<Integer>> originAssignments) { Map<String, Assignment<Integer>> newAssignments = new HashMap<String, Assignment<Integer>>(); if (brokers == null || brokers.isEmpty() || topics == null || topics.isEmpty()) { return newAssignments; } Map<String, List<Pair<String, Integer>>> avalibleOriginAssignments = new HashMap<String, List<Pair<String, Integer>>>(); List<Pair<String, Integer>> freeTps = new ArrayList<Pair<String, Integer>>(); for (String broker : brokers.keySet()) { avalibleOriginAssignments.put(broker, new ArrayList<Pair<String, Integer>>()); } for (Topic topic : topics) { if (originAssignments.containsKey(topic.getName())) { Assignment<Integer> topicOriginAssignments = originAssignments.get(topic.getName()); for (Partition partition : topic.getPartitions()) { Map<String, ClientContext> originTPAssianment = topicOriginAssignments.getAssignment(partition.getId()); if (originTPAssianment != null) { if (originTPAssianment.size() != 1) { log.error("TP have more than one broker assigned"); } ClientContext broker = originTPAssianment.values().iterator().next(); if (brokers.containsKey(broker.getName())) { avalibleOriginAssignments.get(broker.getName()).add( new Pair<String, Integer>(topic.getName(), partition.getId())); continue; } } freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } else { for (Partition partition : topic.getPartitions()) { freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } } Map<String, List<Pair<String, Integer>>> newAssigns = m_assignBalancer.assign(avalibleOriginAssignments, freeTps); for (Entry<String, List<Pair<String, Integer>>> assign : newAssigns.entrySet()) { for (Pair<String, Integer> tp : assign.getValue()) { if (!newAssignments.containsKey(tp.getKey())) { newAssignments.put(tp.getKey(), new Assignment<Integer>()); } Map<String, ClientContext> broker = new HashMap<String, ClientContext>(); broker.put(assign.getKey(), brokers.get(assign.getKey())); newAssignments.get(tp.getKey()).addAssignment(tp.getValue(), broker); } } return newAssignments; } }
LeastAdjustmentBrokerPartitionAssigningStrategy implements BrokerPartitionAssigningStrategy { @Override public Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics, Map<String, Assignment<Integer>> originAssignments) { Map<String, Assignment<Integer>> newAssignments = new HashMap<String, Assignment<Integer>>(); if (brokers == null || brokers.isEmpty() || topics == null || topics.isEmpty()) { return newAssignments; } Map<String, List<Pair<String, Integer>>> avalibleOriginAssignments = new HashMap<String, List<Pair<String, Integer>>>(); List<Pair<String, Integer>> freeTps = new ArrayList<Pair<String, Integer>>(); for (String broker : brokers.keySet()) { avalibleOriginAssignments.put(broker, new ArrayList<Pair<String, Integer>>()); } for (Topic topic : topics) { if (originAssignments.containsKey(topic.getName())) { Assignment<Integer> topicOriginAssignments = originAssignments.get(topic.getName()); for (Partition partition : topic.getPartitions()) { Map<String, ClientContext> originTPAssianment = topicOriginAssignments.getAssignment(partition.getId()); if (originTPAssianment != null) { if (originTPAssianment.size() != 1) { log.error("TP have more than one broker assigned"); } ClientContext broker = originTPAssianment.values().iterator().next(); if (brokers.containsKey(broker.getName())) { avalibleOriginAssignments.get(broker.getName()).add( new Pair<String, Integer>(topic.getName(), partition.getId())); continue; } } freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } else { for (Partition partition : topic.getPartitions()) { freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } } Map<String, List<Pair<String, Integer>>> newAssigns = m_assignBalancer.assign(avalibleOriginAssignments, freeTps); for (Entry<String, List<Pair<String, Integer>>> assign : newAssigns.entrySet()) { for (Pair<String, Integer> tp : assign.getValue()) { if (!newAssignments.containsKey(tp.getKey())) { newAssignments.put(tp.getKey(), new Assignment<Integer>()); } Map<String, ClientContext> broker = new HashMap<String, ClientContext>(); broker.put(assign.getKey(), brokers.get(assign.getKey())); newAssignments.get(tp.getKey()).addAssignment(tp.getValue(), broker); } } return newAssignments; }  }
LeastAdjustmentBrokerPartitionAssigningStrategy implements BrokerPartitionAssigningStrategy { @Override public Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics, Map<String, Assignment<Integer>> originAssignments) { Map<String, Assignment<Integer>> newAssignments = new HashMap<String, Assignment<Integer>>(); if (brokers == null || brokers.isEmpty() || topics == null || topics.isEmpty()) { return newAssignments; } Map<String, List<Pair<String, Integer>>> avalibleOriginAssignments = new HashMap<String, List<Pair<String, Integer>>>(); List<Pair<String, Integer>> freeTps = new ArrayList<Pair<String, Integer>>(); for (String broker : brokers.keySet()) { avalibleOriginAssignments.put(broker, new ArrayList<Pair<String, Integer>>()); } for (Topic topic : topics) { if (originAssignments.containsKey(topic.getName())) { Assignment<Integer> topicOriginAssignments = originAssignments.get(topic.getName()); for (Partition partition : topic.getPartitions()) { Map<String, ClientContext> originTPAssianment = topicOriginAssignments.getAssignment(partition.getId()); if (originTPAssianment != null) { if (originTPAssianment.size() != 1) { log.error("TP have more than one broker assigned"); } ClientContext broker = originTPAssianment.values().iterator().next(); if (brokers.containsKey(broker.getName())) { avalibleOriginAssignments.get(broker.getName()).add( new Pair<String, Integer>(topic.getName(), partition.getId())); continue; } } freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } else { for (Partition partition : topic.getPartitions()) { freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } } Map<String, List<Pair<String, Integer>>> newAssigns = m_assignBalancer.assign(avalibleOriginAssignments, freeTps); for (Entry<String, List<Pair<String, Integer>>> assign : newAssigns.entrySet()) { for (Pair<String, Integer> tp : assign.getValue()) { if (!newAssignments.containsKey(tp.getKey())) { newAssignments.put(tp.getKey(), new Assignment<Integer>()); } Map<String, ClientContext> broker = new HashMap<String, ClientContext>(); broker.put(assign.getKey(), brokers.get(assign.getKey())); newAssignments.get(tp.getKey()).addAssignment(tp.getValue(), broker); } } return newAssignments; }  @Override Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics,
	      Map<String, Assignment<Integer>> originAssignments); }
LeastAdjustmentBrokerPartitionAssigningStrategy implements BrokerPartitionAssigningStrategy { @Override public Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics, Map<String, Assignment<Integer>> originAssignments) { Map<String, Assignment<Integer>> newAssignments = new HashMap<String, Assignment<Integer>>(); if (brokers == null || brokers.isEmpty() || topics == null || topics.isEmpty()) { return newAssignments; } Map<String, List<Pair<String, Integer>>> avalibleOriginAssignments = new HashMap<String, List<Pair<String, Integer>>>(); List<Pair<String, Integer>> freeTps = new ArrayList<Pair<String, Integer>>(); for (String broker : brokers.keySet()) { avalibleOriginAssignments.put(broker, new ArrayList<Pair<String, Integer>>()); } for (Topic topic : topics) { if (originAssignments.containsKey(topic.getName())) { Assignment<Integer> topicOriginAssignments = originAssignments.get(topic.getName()); for (Partition partition : topic.getPartitions()) { Map<String, ClientContext> originTPAssianment = topicOriginAssignments.getAssignment(partition.getId()); if (originTPAssianment != null) { if (originTPAssianment.size() != 1) { log.error("TP have more than one broker assigned"); } ClientContext broker = originTPAssianment.values().iterator().next(); if (brokers.containsKey(broker.getName())) { avalibleOriginAssignments.get(broker.getName()).add( new Pair<String, Integer>(topic.getName(), partition.getId())); continue; } } freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } else { for (Partition partition : topic.getPartitions()) { freeTps.add(new Pair<String, Integer>(topic.getName(), partition.getId())); } } } Map<String, List<Pair<String, Integer>>> newAssigns = m_assignBalancer.assign(avalibleOriginAssignments, freeTps); for (Entry<String, List<Pair<String, Integer>>> assign : newAssigns.entrySet()) { for (Pair<String, Integer> tp : assign.getValue()) { if (!newAssignments.containsKey(tp.getKey())) { newAssignments.put(tp.getKey(), new Assignment<Integer>()); } Map<String, ClientContext> broker = new HashMap<String, ClientContext>(); broker.put(assign.getKey(), brokers.get(assign.getKey())); newAssignments.get(tp.getKey()).addAssignment(tp.getValue(), broker); } } return newAssignments; }  @Override Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics,
	      Map<String, Assignment<Integer>> originAssignments);  }