@Test public void test() throws Exception { long now = System.currentTimeMillis(); Map<String, Assignment<Integer>> brokerAssignments = new HashMap<>(); Assignment<Integer> t1Assignment = new Assignment<Integer>(); t1Assignment.addAssignment(1, createBroker("br1", "1.1.1.1", 1111)); t1Assignment.addAssignment(2, createBroker("br2", "1.1.1.2", 2222)); t1Assignment.addAssignment(3, createBroker("br1", "1.1.1.1", 1111)); brokerAssignments.put("t1", t1Assignment); Assignment<Integer> t2Assignment = new Assignment<Integer>(); t2Assignment.addAssignment(1, createBroker("br1", "1.1.1.1", 1111)); t2Assignment.addAssignment(2, createBroker("br2", "1.1.1.2", 2222)); t2Assignment.addAssignment(3, createBroker("br3", "1.1.1.3", 3333)); brokerAssignments.put("t2", t2Assignment); Map<Pair<String, Integer>, Map<String, ClientLeaseInfo>> leases = new HashMap<>(); leases.put(new Pair<String, Integer>("t1", 2), createBrokerLease("br1", "1.1.1.11", 8888, 1, now + 10 * 1000L)); leases.put(new Pair<String, Integer>("t1", 4), createBrokerLease("br1", "1.1.1.11", 8888, 1, now + 10 * 1000L)); leases.put(new Pair<String, Integer>("t2", 1), createBrokerLease("br1", "1.1.1.1", 1111, 2, now + 20 * 1000L)); when(m_brokerLeaseHolder.getAllValidLeases()).thenReturn(leases); when(m_scheduledExecutor.schedule(any(Runnable.class), anyLong(), eq(TimeUnit.MILLISECONDS))).thenReturn(null); Map<String, Map<Integer, Endpoint>> endpoints = m_maker.makeEndpoints(null, -1, null, brokerAssignments, false); verify(m_scheduledExecutor, times(1)).schedule(any(Runnable.class), anyLong(), eq(TimeUnit.MILLISECONDS)); assertEquals(2, endpoints.size()); Map<Integer, Endpoint> endpoints1 = endpoints.get("t1"); assertEquals(3, endpoints1.size()); assertEndpoint("br1", "1.1.1.1", 1111, endpoints1.get(1)); assertEndpoint("br1", "1.1.1.11", 8888, endpoints1.get(2)); assertEndpoint("br1", "1.1.1.1", 1111, endpoints1.get(3)); Map<Integer, Endpoint> endpoints2 = endpoints.get("t2"); assertEquals(3, endpoints2.size()); assertEndpoint("br1", "1.1.1.1", 1111, endpoints2.get(1)); assertEndpoint("br2", "1.1.1.2", 2222, endpoints2.get(2)); assertEndpoint("br3", "1.1.1.3", 3333, endpoints2.get(3)); }
public Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version, ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce) throws Exception { long start = System.currentTimeMillis(); try { Map<String, Map<Integer, Endpoint>> topicPartition2Endpoints = new HashMap<>(); Pair<Long, Long> delayRebalanceTimespan = new Pair<>(0L, 0L); Map<Pair<String, Integer>, Map<String, ClientLeaseInfo>> brokerLeases = m_brokerLeaseHolder .getAllValidLeases(); for (Map.Entry<String, Assignment<Integer>> topicAssignment : brokerAssignments.entrySet()) { String topicName = topicAssignment.getKey(); Map<Integer, Map<String, ClientContext>> assignment = topicAssignment.getValue().getAssignments(); if (assignment != null && !assignment.isEmpty()) { topicPartition2Endpoints.put(topicName, new HashMap<Integer, Endpoint>()); for (Map.Entry<Integer, Map<String, ClientContext>> partitionAssignment : assignment.entrySet()) { int partition = partitionAssignment.getKey(); Map<String, ClientContext> assignedBrokers = partitionAssignment.getValue(); topicPartition2Endpoints.get(topicName).putAll( makePartition2Endpoints(topicName, partition, assignedBrokers, brokerLeases.get(new Pair<>(topicName, partition)), delayRebalanceTimespan)); } } } if (!mergeOnce) { scheduleDelayReblanceTasks(eventBus, version, stateHolder, delayRebalanceTimespan); } return topicPartition2Endpoints; } finally { log.info("Make endpoint cost {}ms", (System.currentTimeMillis() - start)); } }
EndpointMaker implements Initializable { public Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version, ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce) throws Exception { long start = System.currentTimeMillis(); try { Map<String, Map<Integer, Endpoint>> topicPartition2Endpoints = new HashMap<>(); Pair<Long, Long> delayRebalanceTimespan = new Pair<>(0L, 0L); Map<Pair<String, Integer>, Map<String, ClientLeaseInfo>> brokerLeases = m_brokerLeaseHolder .getAllValidLeases(); for (Map.Entry<String, Assignment<Integer>> topicAssignment : brokerAssignments.entrySet()) { String topicName = topicAssignment.getKey(); Map<Integer, Map<String, ClientContext>> assignment = topicAssignment.getValue().getAssignments(); if (assignment != null && !assignment.isEmpty()) { topicPartition2Endpoints.put(topicName, new HashMap<Integer, Endpoint>()); for (Map.Entry<Integer, Map<String, ClientContext>> partitionAssignment : assignment.entrySet()) { int partition = partitionAssignment.getKey(); Map<String, ClientContext> assignedBrokers = partitionAssignment.getValue(); topicPartition2Endpoints.get(topicName).putAll( makePartition2Endpoints(topicName, partition, assignedBrokers, brokerLeases.get(new Pair<>(topicName, partition)), delayRebalanceTimespan)); } } } if (!mergeOnce) { scheduleDelayReblanceTasks(eventBus, version, stateHolder, delayRebalanceTimespan); } return topicPartition2Endpoints; } finally { log.info("Make endpoint cost {}ms", (System.currentTimeMillis() - start)); } } }
EndpointMaker implements Initializable { public Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version, ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce) throws Exception { long start = System.currentTimeMillis(); try { Map<String, Map<Integer, Endpoint>> topicPartition2Endpoints = new HashMap<>(); Pair<Long, Long> delayRebalanceTimespan = new Pair<>(0L, 0L); Map<Pair<String, Integer>, Map<String, ClientLeaseInfo>> brokerLeases = m_brokerLeaseHolder .getAllValidLeases(); for (Map.Entry<String, Assignment<Integer>> topicAssignment : brokerAssignments.entrySet()) { String topicName = topicAssignment.getKey(); Map<Integer, Map<String, ClientContext>> assignment = topicAssignment.getValue().getAssignments(); if (assignment != null && !assignment.isEmpty()) { topicPartition2Endpoints.put(topicName, new HashMap<Integer, Endpoint>()); for (Map.Entry<Integer, Map<String, ClientContext>> partitionAssignment : assignment.entrySet()) { int partition = partitionAssignment.getKey(); Map<String, ClientContext> assignedBrokers = partitionAssignment.getValue(); topicPartition2Endpoints.get(topicName).putAll( makePartition2Endpoints(topicName, partition, assignedBrokers, brokerLeases.get(new Pair<>(topicName, partition)), delayRebalanceTimespan)); } } } if (!mergeOnce) { scheduleDelayReblanceTasks(eventBus, version, stateHolder, delayRebalanceTimespan); } return topicPartition2Endpoints; } finally { log.info("Make endpoint cost {}ms", (System.currentTimeMillis() - start)); } }  }
EndpointMaker implements Initializable { public Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version, ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce) throws Exception { long start = System.currentTimeMillis(); try { Map<String, Map<Integer, Endpoint>> topicPartition2Endpoints = new HashMap<>(); Pair<Long, Long> delayRebalanceTimespan = new Pair<>(0L, 0L); Map<Pair<String, Integer>, Map<String, ClientLeaseInfo>> brokerLeases = m_brokerLeaseHolder .getAllValidLeases(); for (Map.Entry<String, Assignment<Integer>> topicAssignment : brokerAssignments.entrySet()) { String topicName = topicAssignment.getKey(); Map<Integer, Map<String, ClientContext>> assignment = topicAssignment.getValue().getAssignments(); if (assignment != null && !assignment.isEmpty()) { topicPartition2Endpoints.put(topicName, new HashMap<Integer, Endpoint>()); for (Map.Entry<Integer, Map<String, ClientContext>> partitionAssignment : assignment.entrySet()) { int partition = partitionAssignment.getKey(); Map<String, ClientContext> assignedBrokers = partitionAssignment.getValue(); topicPartition2Endpoints.get(topicName).putAll( makePartition2Endpoints(topicName, partition, assignedBrokers, brokerLeases.get(new Pair<>(topicName, partition)), delayRebalanceTimespan)); } } } if (!mergeOnce) { scheduleDelayReblanceTasks(eventBus, version, stateHolder, delayRebalanceTimespan); } return topicPartition2Endpoints; } finally { log.info("Make endpoint cost {}ms", (System.currentTimeMillis() - start)); } }  void setBrokerLeaseHolder(BrokerLeaseHolder brokerLeaseHolder); void setConfig(MetaServerConfig config); void setScheduledExecutor(ScheduledExecutorService scheduledExecutor); Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version,
	      ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce); @Override void initialize(); }
EndpointMaker implements Initializable { public Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version, ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce) throws Exception { long start = System.currentTimeMillis(); try { Map<String, Map<Integer, Endpoint>> topicPartition2Endpoints = new HashMap<>(); Pair<Long, Long> delayRebalanceTimespan = new Pair<>(0L, 0L); Map<Pair<String, Integer>, Map<String, ClientLeaseInfo>> brokerLeases = m_brokerLeaseHolder .getAllValidLeases(); for (Map.Entry<String, Assignment<Integer>> topicAssignment : brokerAssignments.entrySet()) { String topicName = topicAssignment.getKey(); Map<Integer, Map<String, ClientContext>> assignment = topicAssignment.getValue().getAssignments(); if (assignment != null && !assignment.isEmpty()) { topicPartition2Endpoints.put(topicName, new HashMap<Integer, Endpoint>()); for (Map.Entry<Integer, Map<String, ClientContext>> partitionAssignment : assignment.entrySet()) { int partition = partitionAssignment.getKey(); Map<String, ClientContext> assignedBrokers = partitionAssignment.getValue(); topicPartition2Endpoints.get(topicName).putAll( makePartition2Endpoints(topicName, partition, assignedBrokers, brokerLeases.get(new Pair<>(topicName, partition)), delayRebalanceTimespan)); } } } if (!mergeOnce) { scheduleDelayReblanceTasks(eventBus, version, stateHolder, delayRebalanceTimespan); } return topicPartition2Endpoints; } finally { log.info("Make endpoint cost {}ms", (System.currentTimeMillis() - start)); } }  void setBrokerLeaseHolder(BrokerLeaseHolder brokerLeaseHolder); void setConfig(MetaServerConfig config); void setScheduledExecutor(ScheduledExecutorService scheduledExecutor); Map<String, Map<Integer, Endpoint>> makeEndpoints(EventBus eventBus, long version,
	      ClusterStateHolder stateHolder, Map<String, Assignment<Integer>> brokerAssignments, boolean mergeOnce); @Override void initialize();  }