@Test public void testMaxSend() { ClusterStatusPublisher csp = new ClusterStatusPublisher() { @Override protected List<Pair<ServerName, Long>> getDeadServers(long since) { List<Pair<ServerName, Long>> res = new ArrayList<Pair<ServerName, Long>>(); switch ((int) EnvironmentEdgeManager.currentTimeMillis()) { case 2: res.add(new Pair<ServerName, Long>(ServerName.valueOf("hn", 10, 10), 1L)); break; case 1000: break; } return res; } }; mee.setValue(2); for (int i = 0; i < ClusterStatusPublisher.NB_SEND; i++) { Assert.assertEquals("i=" + i, 1, csp.generateDeadServersListToSend().size()); } mee.setValue(1000); Assert.assertTrue(csp.generateDeadServersListToSend().isEmpty()); }
protected List<ServerName> generateDeadServersListToSend() { long since = EnvironmentEdgeManager.currentTimeMillis() - messagePeriod * 2; for (Pair<ServerName, Long> dead : getDeadServers(since)) { lastSent.putIfAbsent(dead.getFirst(), 0); } List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>(); entries.addAll(lastSent.entrySet()); Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() { @Override public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) { return o1.getValue().compareTo(o2.getValue()); } }); int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size(); List<ServerName> res = new ArrayList<ServerName>(max); for (int i = 0; i < max; i++) { Map.Entry<ServerName, Integer> toSend = entries.get(i); if (toSend.getValue() >= (NB_SEND - 1)) { lastSent.remove(toSend.getKey()); } else { lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1); } res.add(toSend.getKey()); } return res; }
ClusterStatusPublisher extends Chore { protected List<ServerName> generateDeadServersListToSend() { long since = EnvironmentEdgeManager.currentTimeMillis() - messagePeriod * 2; for (Pair<ServerName, Long> dead : getDeadServers(since)) { lastSent.putIfAbsent(dead.getFirst(), 0); } List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>(); entries.addAll(lastSent.entrySet()); Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() { @Override public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) { return o1.getValue().compareTo(o2.getValue()); } }); int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size(); List<ServerName> res = new ArrayList<ServerName>(max); for (int i = 0; i < max; i++) { Map.Entry<ServerName, Integer> toSend = entries.get(i); if (toSend.getValue() >= (NB_SEND - 1)) { lastSent.remove(toSend.getKey()); } else { lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1); } res.add(toSend.getKey()); } return res; } }
ClusterStatusPublisher extends Chore { protected List<ServerName> generateDeadServersListToSend() { long since = EnvironmentEdgeManager.currentTimeMillis() - messagePeriod * 2; for (Pair<ServerName, Long> dead : getDeadServers(since)) { lastSent.putIfAbsent(dead.getFirst(), 0); } List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>(); entries.addAll(lastSent.entrySet()); Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() { @Override public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) { return o1.getValue().compareTo(o2.getValue()); } }); int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size(); List<ServerName> res = new ArrayList<ServerName>(max); for (int i = 0; i < max; i++) { Map.Entry<ServerName, Integer> toSend = entries.get(i); if (toSend.getValue() >= (NB_SEND - 1)) { lastSent.remove(toSend.getKey()); } else { lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1); } res.add(toSend.getKey()); } return res; } ClusterStatusPublisher(HMaster master, Configuration conf,
                                Class<? extends Publisher> publisherClass); protected  ClusterStatusPublisher(); }
ClusterStatusPublisher extends Chore { protected List<ServerName> generateDeadServersListToSend() { long since = EnvironmentEdgeManager.currentTimeMillis() - messagePeriod * 2; for (Pair<ServerName, Long> dead : getDeadServers(since)) { lastSent.putIfAbsent(dead.getFirst(), 0); } List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>(); entries.addAll(lastSent.entrySet()); Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() { @Override public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) { return o1.getValue().compareTo(o2.getValue()); } }); int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size(); List<ServerName> res = new ArrayList<ServerName>(max); for (int i = 0; i < max; i++) { Map.Entry<ServerName, Integer> toSend = entries.get(i); if (toSend.getValue() >= (NB_SEND - 1)) { lastSent.remove(toSend.getKey()); } else { lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1); } res.add(toSend.getKey()); } return res; } ClusterStatusPublisher(HMaster master, Configuration conf,
                                Class<? extends Publisher> publisherClass); protected  ClusterStatusPublisher();  }
ClusterStatusPublisher extends Chore { protected List<ServerName> generateDeadServersListToSend() { long since = EnvironmentEdgeManager.currentTimeMillis() - messagePeriod * 2; for (Pair<ServerName, Long> dead : getDeadServers(since)) { lastSent.putIfAbsent(dead.getFirst(), 0); } List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>(); entries.addAll(lastSent.entrySet()); Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() { @Override public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) { return o1.getValue().compareTo(o2.getValue()); } }); int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size(); List<ServerName> res = new ArrayList<ServerName>(max); for (int i = 0; i < max; i++) { Map.Entry<ServerName, Integer> toSend = entries.get(i); if (toSend.getValue() >= (NB_SEND - 1)) { lastSent.remove(toSend.getKey()); } else { lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1); } res.add(toSend.getKey()); } return res; } ClusterStatusPublisher(HMaster master, Configuration conf,
                                Class<? extends Publisher> publisherClass); protected  ClusterStatusPublisher();  static final String STATUS_PUBLISHER_CLASS; static final Class<? extends ClusterStatusPublisher.Publisher> DEFAULT_STATUS_PUBLISHER_CLASS; static final String STATUS_PUBLISH_PERIOD; static final int DEFAULT_STATUS_PUBLISH_PERIOD; final static int MAX_SERVER_PER_MESSAGE; final static int NB_SEND; }