SlowPeerTracker { public void addReport(String slowNode, String reportingNode) { ConcurrentMap<String, Long> nodeEntries = allReports.get(slowNode); if (nodeEntries == null) { allReports.putIfAbsent(slowNode, new ConcurrentHashMap<>()); nodeEntries = allReports.get(slowNode); } nodeEntries.put(reportingNode, timer.monotonicNow()); } SlowPeerTracker(Configuration conf, Timer timer); void addReport(String slowNode,
                        String reportingNode); Set<String> getReportsForNode(String slowNode); Map<String, SortedSet<String>> getReportsForAllDataNodes(); String getJson(); static final Logger LOG; }
@Test public void testGetJsonSizeIsLimited() throws IOException { tracker.addReport("node1", "node2"); tracker.addReport("node1", "node3"); tracker.addReport("node2", "node3"); tracker.addReport("node2", "node4"); tracker.addReport("node3", "node4"); tracker.addReport("node3", "node5"); tracker.addReport("node4", "node6"); tracker.addReport("node5", "node6"); tracker.addReport("node5", "node7"); tracker.addReport("node6", "node7"); tracker.addReport("node6", "node8"); final Set<ReportForJson> reports = getAndDeserializeJson(); assertFalse(isNodeInReports(reports, "node4")); assertTrue(isNodeInReports(reports, "node1")); assertTrue(isNodeInReports(reports, "node2")); assertTrue(isNodeInReports(reports, "node3")); assertTrue(isNodeInReports(reports, "node5")); assertTrue(isNodeInReports(reports, "node6")); }
@Test public void testLowRankedElementsIgnored() throws IOException { for (int i = 0; i < 5; ++i) { tracker.addReport("node" + i, "reporter1"); tracker.addReport("node" + i, "reporter2"); } for (int i = 10; i < 20; ++i) { tracker.addReport("node" + i, "reporter1"); } final Set<ReportForJson> reports = getAndDeserializeJson(); for (int i = 0; i < 5; ++i) { assertTrue(isNodeInReports(reports, "node" + i)); } }