@Test public void test() throws Exception { Map<Pair<String, String>, Map<String, ClientContext>> changes1 = new HashMap<>(); Pair<String, String> t1g1 = new Pair<String, String>("t1", "g1"); Map<String, ClientContext> t1g1Consumers = new LinkedHashMap<>(); t1g1Consumers.put("c1", new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L)); t1g1Consumers.put("c2", new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L)); changes1.put(t1g1, t1g1Consumers); Pair<String, String> t1g4 = new Pair<String, String>("t1", "g4"); Map<String, ClientContext> t1g4Consumers = new LinkedHashMap<>(); t1g4Consumers.put("c1", new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L)); changes1.put(t1g4, t1g4Consumers); Pair<String, String> t1g2 = new Pair<String, String>("t1", "g2"); Map<String, ClientContext> t1g2Consumers = new LinkedHashMap<>(); t1g2Consumers.put("c3", new ClientContext("c3", "3.3.3.3", 3333, null, null, -3L)); t1g2Consumers.put("c4", new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L)); changes1.put(t1g2, t1g2Consumers); Pair<String, String> t2g1 = new Pair<String, String>("t2", "g1"); Map<String, ClientContext> t2g1Consumers = new LinkedHashMap<>(); t2g1Consumers.put("c1", new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L)); t2g1Consumers.put("c4", new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L)); changes1.put(t2g1, t2g1Consumers); Pair<String, String> t2g2 = new Pair<String, String>("t2", "g2"); Map<String, ClientContext> t2g2Consumers = new LinkedHashMap<>(); t2g2Consumers.put("c2", new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L)); t2g2Consumers.put("c3", new ClientContext("c3", "3.3.3.3", 3333, null, null, -3L)); changes1.put(t2g2, t2g2Consumers); Pair<String, String> t3g1 = new Pair<String, String>("t3", "g1"); Map<String, ClientContext> t3g1Consumers = new LinkedHashMap<>(); t3g1Consumers.put("c2", new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L)); changes1.put(t3g1, t3g1Consumers); when(m_activeConsumerListHolder.scanChanges(anyLong(), any(TimeUnit.class))).thenReturn(changes1); m_holder.rebalance(); assertAssignment(t1g1, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L))) ) ); assertAssignment(t1g2, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c3", "3.3.3.3", 3333, null, null, -3L), new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c3", "3.3.3.3", 3333, null, null, -3L), new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))) ) ); assertAssignment(t2g1, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L), new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L), new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))) ) ); assertAssignment(t2g2, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c3", "3.3.3.3", 3333, null, null, -3L))) ) ); Map<Pair<String, String>, Map<String, ClientContext>> changes2 = new HashMap<>(changes1); changes2.get(t1g1).remove("c1"); changes2.get(t1g2).remove("c3"); changes2.get(t2g2).remove("c2"); changes2.get(t2g2).put("c4", new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L)); reset(m_activeConsumerListHolder); when(m_activeConsumerListHolder.scanChanges(anyLong(), any(TimeUnit.class))).thenReturn(changes2); m_holder.rebalance(); assertAssignment(t1g1, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c2", "2.2.2.2", 2222, null, null, -2L))) ) ); assertAssignment(t1g2, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))) ) ); assertAssignment(t2g1, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L), new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c1", "1.1.1.1", 1111, null, null, -1L), new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))) ) ); assertAssignment(t2g2, Arrays.asList( new Pair<Integer, List<ClientContext>>(0, Arrays.asList(new ClientContext("c3", "3.3.3.3", 3333, null, null, -3L))), new Pair<Integer, List<ClientContext>>(1, Arrays.asList(new ClientContext("c4", "4.4.4.4", 4444, null, null, -4L))) ) ); }
protected void rebalance() { try { Map<Pair<String, String>, Map<String, ClientContext>> changes = m_activeConsumerListHolder.scanChanges( m_config.getConsumerHeartbeatTimeoutMillis(), TimeUnit.MILLISECONDS); if (changes != null && !changes.isEmpty()) { Map<Pair<String, String>, Assignment<Integer>> newAssignments = new HashMap<>(m_assignments.get()); for (Map.Entry<Pair<String, String>, Map<String, ClientContext>> change : changes.entrySet()) { Pair<String, String> topicGroup = change.getKey(); Map<String, ClientContext> consumerList = change.getValue(); if (consumerList == null || consumerList.isEmpty()) { newAssignments.remove(topicGroup); } else { Assignment<Integer> newAssignment = createNewAssignment(topicGroup, consumerList, newAssignments.get(topicGroup)); if (newAssignment != null) { newAssignments.put(topicGroup, newAssignment); } } } m_assignments.set(newAssignments); if (traceLog.isInfoEnabled()) { traceLog.info("Consumer assignment changed.\n{}", JSON.toJSONString(newAssignments)); } } } catch (Exception e) { log.warn("Error occurred while doing assignment check in ConsumerRebalanceChecker", e); } }
ConsumerAssignmentHolder implements Initializable { protected void rebalance() { try { Map<Pair<String, String>, Map<String, ClientContext>> changes = m_activeConsumerListHolder.scanChanges( m_config.getConsumerHeartbeatTimeoutMillis(), TimeUnit.MILLISECONDS); if (changes != null && !changes.isEmpty()) { Map<Pair<String, String>, Assignment<Integer>> newAssignments = new HashMap<>(m_assignments.get()); for (Map.Entry<Pair<String, String>, Map<String, ClientContext>> change : changes.entrySet()) { Pair<String, String> topicGroup = change.getKey(); Map<String, ClientContext> consumerList = change.getValue(); if (consumerList == null || consumerList.isEmpty()) { newAssignments.remove(topicGroup); } else { Assignment<Integer> newAssignment = createNewAssignment(topicGroup, consumerList, newAssignments.get(topicGroup)); if (newAssignment != null) { newAssignments.put(topicGroup, newAssignment); } } } m_assignments.set(newAssignments); if (traceLog.isInfoEnabled()) { traceLog.info("Consumer assignment changed.\n{}", JSON.toJSONString(newAssignments)); } } } catch (Exception e) { log.warn("Error occurred while doing assignment check in ConsumerRebalanceChecker", e); } } }
ConsumerAssignmentHolder implements Initializable { protected void rebalance() { try { Map<Pair<String, String>, Map<String, ClientContext>> changes = m_activeConsumerListHolder.scanChanges( m_config.getConsumerHeartbeatTimeoutMillis(), TimeUnit.MILLISECONDS); if (changes != null && !changes.isEmpty()) { Map<Pair<String, String>, Assignment<Integer>> newAssignments = new HashMap<>(m_assignments.get()); for (Map.Entry<Pair<String, String>, Map<String, ClientContext>> change : changes.entrySet()) { Pair<String, String> topicGroup = change.getKey(); Map<String, ClientContext> consumerList = change.getValue(); if (consumerList == null || consumerList.isEmpty()) { newAssignments.remove(topicGroup); } else { Assignment<Integer> newAssignment = createNewAssignment(topicGroup, consumerList, newAssignments.get(topicGroup)); if (newAssignment != null) { newAssignments.put(topicGroup, newAssignment); } } } m_assignments.set(newAssignments); if (traceLog.isInfoEnabled()) { traceLog.info("Consumer assignment changed.\n{}", JSON.toJSONString(newAssignments)); } } } catch (Exception e) { log.warn("Error occurred while doing assignment check in ConsumerRebalanceChecker", e); } } ConsumerAssignmentHolder(); }
ConsumerAssignmentHolder implements Initializable { protected void rebalance() { try { Map<Pair<String, String>, Map<String, ClientContext>> changes = m_activeConsumerListHolder.scanChanges( m_config.getConsumerHeartbeatTimeoutMillis(), TimeUnit.MILLISECONDS); if (changes != null && !changes.isEmpty()) { Map<Pair<String, String>, Assignment<Integer>> newAssignments = new HashMap<>(m_assignments.get()); for (Map.Entry<Pair<String, String>, Map<String, ClientContext>> change : changes.entrySet()) { Pair<String, String> topicGroup = change.getKey(); Map<String, ClientContext> consumerList = change.getValue(); if (consumerList == null || consumerList.isEmpty()) { newAssignments.remove(topicGroup); } else { Assignment<Integer> newAssignment = createNewAssignment(topicGroup, consumerList, newAssignments.get(topicGroup)); if (newAssignment != null) { newAssignments.put(topicGroup, newAssignment); } } } m_assignments.set(newAssignments); if (traceLog.isInfoEnabled()) { traceLog.info("Consumer assignment changed.\n{}", JSON.toJSONString(newAssignments)); } } } catch (Exception e) { log.warn("Error occurred while doing assignment check in ConsumerRebalanceChecker", e); } } ConsumerAssignmentHolder(); void setActiveConsumerListHolder(ActiveConsumerListHolder activeConsumerListHolder); void setConfig(MetaServerConfig config); void setMetaHolder(MetaHolder metaHolder); void setPartitionAssigningStrategy(ConsumerPartitionAssigningStrategy partitionAssigningStrategy); Assignment<Integer> getAssignment(Pair<String, String> topicGroup); Map<Pair<String, String>, Assignment<Integer>> getAssignments(); @Override void initialize(); }
ConsumerAssignmentHolder implements Initializable { protected void rebalance() { try { Map<Pair<String, String>, Map<String, ClientContext>> changes = m_activeConsumerListHolder.scanChanges( m_config.getConsumerHeartbeatTimeoutMillis(), TimeUnit.MILLISECONDS); if (changes != null && !changes.isEmpty()) { Map<Pair<String, String>, Assignment<Integer>> newAssignments = new HashMap<>(m_assignments.get()); for (Map.Entry<Pair<String, String>, Map<String, ClientContext>> change : changes.entrySet()) { Pair<String, String> topicGroup = change.getKey(); Map<String, ClientContext> consumerList = change.getValue(); if (consumerList == null || consumerList.isEmpty()) { newAssignments.remove(topicGroup); } else { Assignment<Integer> newAssignment = createNewAssignment(topicGroup, consumerList, newAssignments.get(topicGroup)); if (newAssignment != null) { newAssignments.put(topicGroup, newAssignment); } } } m_assignments.set(newAssignments); if (traceLog.isInfoEnabled()) { traceLog.info("Consumer assignment changed.\n{}", JSON.toJSONString(newAssignments)); } } } catch (Exception e) { log.warn("Error occurred while doing assignment check in ConsumerRebalanceChecker", e); } } ConsumerAssignmentHolder(); void setActiveConsumerListHolder(ActiveConsumerListHolder activeConsumerListHolder); void setConfig(MetaServerConfig config); void setMetaHolder(MetaHolder metaHolder); void setPartitionAssigningStrategy(ConsumerPartitionAssigningStrategy partitionAssigningStrategy); Assignment<Integer> getAssignment(Pair<String, String> topicGroup); Map<Pair<String, String>, Assignment<Integer>> getAssignments(); @Override void initialize();  }