WorkloadInfoRetriever { public TopicWorkload topicWorkload(String topic) { LinkedList<TopicWorkload> tws = _topicWorkloadMap.get(topic); if (tws == null || tws.isEmpty()) { return _defaultTopicWorkload; } TopicWorkload maxTw = null; long current = System.currentTimeMillis(); long lookbackWindow = _maxValidTimeMillis; if (tws.stream().anyMatch(topicWorkload -> { return current - topicWorkload.getLastUpdate() < _estimationLookBackWindow; })) { lookbackWindow = _estimationLookBackWindow; } for (TopicWorkload tw : tws) { if (current - tw.getLastUpdate() > lookbackWindow) { continue; } if (maxTw == null || maxTw.getBytesPerSecond() < tw.getBytesPerSecond()) { maxTw = tw; } } return (maxTw != null) ? maxTw : _defaultTopicWorkload; } @VisibleForTesting  WorkloadInfoRetriever();  WorkloadInfoRetriever(IHelixManager helixMirrorMakerManager, boolean isController);  WorkloadInfoRetriever(IHelixManager helixMirrorMakerManager, boolean isController, String srcKafkaZkPath); void start(); void stop(); boolean isInitialized(); TopicWorkload topicWorkload(String topic); void setTopicDefaultWorkload(TopicWorkload defaultWorkload); void refreshWorkloads(); void initializeWorkloads();  }
@Test public void testGetTopicWorkload() { WorkloadInfoRetriever retriever = new WorkloadInfoRetriever(); LinkedList<TopicWorkload> workloadLst = new LinkedList<>(); retriever._topicWorkloadMap.put("trip_created", workloadLst); TopicWorkload wl = new TopicWorkload(10 * 1024, 10); long currentTime = System.currentTimeMillis(); wl.setLastUpdate(currentTime - TimeUnit.HOURS.toMillis(1)); workloadLst.add(wl); TopicWorkload oldWorkload = new TopicWorkload(20 * 1024, 20); oldWorkload.setLastUpdate(currentTime - TimeUnit.HOURS.toMillis(9)); workloadLst.add(oldWorkload); Assert.assertEquals(retriever.topicWorkload("trip_created").compareTo(wl), 0); workloadLst.clear(); workloadLst.add(oldWorkload); Assert.assertEquals(retriever.topicWorkload("trip_created").compareTo(oldWorkload), 0); }