@Test public void testBrokerAdded() throws Exception { long now = System.currentTimeMillis(); Map<String, ClientContext> brokers = createBrokers(Arrays.asList( new ClientContext("br0", "0.0.0.0", 1234, null, null, now), new ClientContext("br1", "0.0.0.1", 1234, null, null, now) )); List<Topic> topics = new ArrayList<>(); topics.add(createTopic("t1", 2, Endpoint.BROKER)); topics.add(createTopic("t2", 2, Endpoint.KAFKA)); topics.add(createTopic("t3", 2, Endpoint.BROKER)); Map<String, Assignment<Integer>> originAssignments = createAssignments(Arrays.asList( new Pair<>(new Pair<>("t1", 0), new ClientContext("br0", "0.0.0.0", 1234, null, null, now)), new Pair<>(new Pair<>("t1", 1), new ClientContext("br0", "0.0.0.0", 1234, null, null, now)), new Pair<>(new Pair<>("t3", 0), new ClientContext("br0", "0.0.0.0", 1234, null, null, now)), new Pair<>(new Pair<>("t3", 1), new ClientContext("br0", "0.0.0.0", 1234, null, null, now)) )); Map<String, Assignment<Integer>> assignments = m_strategy.assign(brokers, topics, originAssignments); assertAssignment(assignments, topics, Arrays.asList( new ClientContext("br0", "0.0.0.0", 1234, null, null, now), new ClientContext("br1", "0.0.0.1", 1234, null, null, now) )); }
@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<>(); if (topics != null && !topics.isEmpty()) { List<Entry<String, ClientContext>> brokerEntries = brokers != null && !brokers.isEmpty() ? new ArrayList<>( brokers.entrySet()) : new ArrayList<Entry<String, ClientContext>>(); int brokerPos = 0; int brokerCount = brokerEntries.size(); for (Topic topic : topics) { if (Endpoint.BROKER.equals(topic.getEndpointType())) { List<Partition> partitions = topic.getPartitions(); if (partitions != null && !partitions.isEmpty()) { Assignment<Integer> assignment = new Assignment<>(); newAssignments.put(topic.getName(), assignment); for (Partition partition : partitions) { Map<String, ClientContext> broker = new HashMap<>(); if (brokerCount > 0) { Entry<String, ClientContext> brokerEntry = brokerEntries.get(brokerPos); brokerPos = (brokerPos + 1) % brokerCount; broker.put(brokerEntry.getKey(), brokerEntry.getValue()); } assignment.addAssignment(partition.getId(), broker); } } } } } return newAssignments; }
DefaultBrokerPartitionAssigningStrategy 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<>(); if (topics != null && !topics.isEmpty()) { List<Entry<String, ClientContext>> brokerEntries = brokers != null && !brokers.isEmpty() ? new ArrayList<>( brokers.entrySet()) : new ArrayList<Entry<String, ClientContext>>(); int brokerPos = 0; int brokerCount = brokerEntries.size(); for (Topic topic : topics) { if (Endpoint.BROKER.equals(topic.getEndpointType())) { List<Partition> partitions = topic.getPartitions(); if (partitions != null && !partitions.isEmpty()) { Assignment<Integer> assignment = new Assignment<>(); newAssignments.put(topic.getName(), assignment); for (Partition partition : partitions) { Map<String, ClientContext> broker = new HashMap<>(); if (brokerCount > 0) { Entry<String, ClientContext> brokerEntry = brokerEntries.get(brokerPos); brokerPos = (brokerPos + 1) % brokerCount; broker.put(brokerEntry.getKey(), brokerEntry.getValue()); } assignment.addAssignment(partition.getId(), broker); } } } } } return newAssignments; } }
DefaultBrokerPartitionAssigningStrategy 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<>(); if (topics != null && !topics.isEmpty()) { List<Entry<String, ClientContext>> brokerEntries = brokers != null && !brokers.isEmpty() ? new ArrayList<>( brokers.entrySet()) : new ArrayList<Entry<String, ClientContext>>(); int brokerPos = 0; int brokerCount = brokerEntries.size(); for (Topic topic : topics) { if (Endpoint.BROKER.equals(topic.getEndpointType())) { List<Partition> partitions = topic.getPartitions(); if (partitions != null && !partitions.isEmpty()) { Assignment<Integer> assignment = new Assignment<>(); newAssignments.put(topic.getName(), assignment); for (Partition partition : partitions) { Map<String, ClientContext> broker = new HashMap<>(); if (brokerCount > 0) { Entry<String, ClientContext> brokerEntry = brokerEntries.get(brokerPos); brokerPos = (brokerPos + 1) % brokerCount; broker.put(brokerEntry.getKey(), brokerEntry.getValue()); } assignment.addAssignment(partition.getId(), broker); } } } } } return newAssignments; }  }
DefaultBrokerPartitionAssigningStrategy 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<>(); if (topics != null && !topics.isEmpty()) { List<Entry<String, ClientContext>> brokerEntries = brokers != null && !brokers.isEmpty() ? new ArrayList<>( brokers.entrySet()) : new ArrayList<Entry<String, ClientContext>>(); int brokerPos = 0; int brokerCount = brokerEntries.size(); for (Topic topic : topics) { if (Endpoint.BROKER.equals(topic.getEndpointType())) { List<Partition> partitions = topic.getPartitions(); if (partitions != null && !partitions.isEmpty()) { Assignment<Integer> assignment = new Assignment<>(); newAssignments.put(topic.getName(), assignment); for (Partition partition : partitions) { Map<String, ClientContext> broker = new HashMap<>(); if (brokerCount > 0) { Entry<String, ClientContext> brokerEntry = brokerEntries.get(brokerPos); brokerPos = (brokerPos + 1) % brokerCount; broker.put(brokerEntry.getKey(), brokerEntry.getValue()); } assignment.addAssignment(partition.getId(), broker); } } } } } return newAssignments; }  @Override Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics,
	      Map<String, Assignment<Integer>> originAssignments); }
DefaultBrokerPartitionAssigningStrategy 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<>(); if (topics != null && !topics.isEmpty()) { List<Entry<String, ClientContext>> brokerEntries = brokers != null && !brokers.isEmpty() ? new ArrayList<>( brokers.entrySet()) : new ArrayList<Entry<String, ClientContext>>(); int brokerPos = 0; int brokerCount = brokerEntries.size(); for (Topic topic : topics) { if (Endpoint.BROKER.equals(topic.getEndpointType())) { List<Partition> partitions = topic.getPartitions(); if (partitions != null && !partitions.isEmpty()) { Assignment<Integer> assignment = new Assignment<>(); newAssignments.put(topic.getName(), assignment); for (Partition partition : partitions) { Map<String, ClientContext> broker = new HashMap<>(); if (brokerCount > 0) { Entry<String, ClientContext> brokerEntry = brokerEntries.get(brokerPos); brokerPos = (brokerPos + 1) % brokerCount; broker.put(brokerEntry.getKey(), brokerEntry.getValue()); } assignment.addAssignment(partition.getId(), broker); } } } } } return newAssignments; }  @Override Map<String, Assignment<Integer>> assign(Map<String, ClientContext> brokers, List<Topic> topics,
	      Map<String, Assignment<Integer>> originAssignments);  }