HelixMirrorMakerManager implements IHelixManager { public TopicPartitionLag calculateLagTime(TopicPartition tp) { TopicPartitionLag tpl = getOffsetMonitor().getTopicPartitionOffset(tp); if (tpl == null || tpl.getLatestOffset() <= 0 || tpl.getCommitOffset() <= 0 || System.currentTimeMillis() - tpl.getTimeStamp() > _offsetMaxValidTimeMillis) { return null; } long lag = tpl.getLatestOffset() - tpl.getCommitOffset(); if (lag <= _minLagOffset) { return null; } double msgRate = getWorkloadInfoRetriever().topicWorkload(tp.getTopic()) .getMsgsPerSecondPerPartition(); if (msgRate < 1) { msgRate = 1; } double lagTime = lag / msgRate; if (lagTime > _minLagTimeSec) { tpl.setLagTime(Math.round(lagTime)); return tpl; } return null; } HelixMirrorMakerManager(ControllerConf controllerConf); double getMaxWorkloadPerWorkerBytes(); double getMaxDedicatedInstancesRatio(); synchronized void start(); synchronized void stop(); synchronized void updateCurrentServingInstance(); synchronized void addTopicToMirrorMaker(TopicPartition topicPartitionInfo); synchronized void addTopicToMirrorMaker(String topicName, int numTopicPartitions); synchronized void expandTopicInMirrorMaker(TopicPartition topicPartitionInfo); synchronized void expandTopicInMirrorMaker(String topicName, int newNumTopicPartitions); Set<TopicPartition> getTopicPartitionBlacklist(); synchronized void updateTopicPartitionStateInMirrorMaker(String topicName, int partition,
      String state); synchronized void deleteTopicInMirrorMaker(String topicName); @Override IdealState getIdealStateForTopic(String topicName); @Override IuReplicatorConf getConf(); ExternalView getExternalViewForTopic(String topicName); boolean isTopicExisted(String topicName); @Override List<String> getTopicLists(); void disableAutoBalancing(); void enableAutoBalancing(); boolean isAutoBalancingEnabled(); boolean isLeader(); List<LiveInstance> getCurrentLiveInstances(); List<String> getCurrentLiveInstanceNames(); void blacklistInstance(String instanceName); void whitelistInstance(String instanceName); List<String> getBlacklistedInstances(); String getHelixZkURL(); String getHelixClusterName(); ControllerConf getControllerConf(); WorkloadInfoRetriever getWorkloadInfoRetriever(); AutoRebalanceLiveInstanceChangeListener getRebalancer(); PriorityQueue<InstanceTopicPartitionHolder> getCurrentServingInstance(); OffsetMonitor getOffsetMonitor(); boolean isHealthy(); TopicPartitionLag calculateLagTime(TopicPartition tp); boolean isSameRegion(String src, String dst);  }
@Test public void testCalculateLagTime() { EasyMock.expect(workloadInfoRetriever.topicWorkload(fakeTopics.get(0))).andReturn(fakeWorkload1).times(3); EasyMock.expect(workloadInfoRetriever.topicWorkload(fakeTopics.get(1))).andReturn(fakeWorkload2).times(2); EasyMock.expect(workloadInfoRetriever.isInitialized()).andReturn(true).anyTimes(); EasyMock.expect(offsetMonitor.getTopicPartitionOffset(fakeTopicPartition1)).andReturn( new TopicPartitionLag(fakeTopicPartition1.getTopic(), fakeTopicPartition1.getPartition(), 12894, 2843)); EasyMock.expect(offsetMonitor.getTopicPartitionOffset(fakeTopicPartition2)).andReturn( new TopicPartitionLag(fakeTopicPartition2.getTopic(), fakeTopicPartition2.getPartition(), 89600093, 2843)); EasyMock.replay(workloadInfoRetriever, offsetMonitor); TopicPartitionLag lag = helixMirrorMakerManager.calculateLagTime(fakeTopicPartition1); Assert.assertNull(lag); TopicPartitionLag lag2 = helixMirrorMakerManager.calculateLagTime(fakeTopicPartition2); Assert.assertEquals(lag2.getLagTime(), 5563); }