@Test(timeout = 10000) public void testRawMetricsCachedAndSourced() throws Exception { ExternalMetricsSink rawMetricsSink = createNiceMock(ExternalMetricsSink.class); expect(rawMetricsSink.getFlushSeconds()).andReturn(2).anyTimes(); expect(rawMetricsSink.getSinkTimeOutSeconds()).andReturn(1).anyTimes(); class CaptureOnce<T> extends Capture<T> { @Override public void setValue(T value) { if (!hasCaptured()) { super.setValue(value); } } } Capture<Collection<TimelineMetrics>> metricsCapture = new CaptureOnce<>(); rawMetricsSink.sinkMetricData(capture(metricsCapture)); expectLastCall(); replay(rawMetricsSink); InternalSourceProvider internalSourceProvider = new DefaultInternalMetricsSourceProvider(); InternalMetricsSource rawMetricsSource = internalSourceProvider.getInternalMetricsSource(RAW_METRICS, 1, rawMetricsSink); TimelineMetrics timelineMetrics = new TimelineMetrics(); final long now = System.currentTimeMillis(); TimelineMetric metric1 = new TimelineMetric(); metric1.setMetricName("m1"); metric1.setAppId("a1"); metric1.setInstanceId("i1"); metric1.setHostName("h1"); metric1.setStartTime(now - 200); metric1.setMetricValues(new TreeMap<Long, Double>() {{ put(now - 100, 1.0); put(now - 200, 2.0); }}); timelineMetrics.getMetrics().add(metric1); rawMetricsSource.publishTimelineMetrics(Collections.singletonList(timelineMetrics)); Thread.sleep(5000); verify(rawMetricsSink); Assert.assertTrue(metricsCapture.hasCaptured()); Assert.assertTrue(metricsCapture.getValue().iterator().next().getMetrics().iterator().next().equals(metric1)); }
@Override public void publishTimelineMetrics(Collection<TimelineMetrics> metrics) { if (rawMetricsSink.getFlushSeconds() > internalCacheInterval) { cache.putAll(metrics); } else { submitDataWithTimeout(metrics); } }
RawMetricsSource implements InternalMetricsSource { @Override public void publishTimelineMetrics(Collection<TimelineMetrics> metrics) { if (rawMetricsSink.getFlushSeconds() > internalCacheInterval) { cache.putAll(metrics); } else { submitDataWithTimeout(metrics); } } }
RawMetricsSource implements InternalMetricsSource { @Override public void publishTimelineMetrics(Collection<TimelineMetrics> metrics) { if (rawMetricsSink.getFlushSeconds() > internalCacheInterval) { cache.putAll(metrics); } else { submitDataWithTimeout(metrics); } } RawMetricsSource(int internalCacheInterval, ExternalMetricsSink rawMetricsSink); }
RawMetricsSource implements InternalMetricsSource { @Override public void publishTimelineMetrics(Collection<TimelineMetrics> metrics) { if (rawMetricsSink.getFlushSeconds() > internalCacheInterval) { cache.putAll(metrics); } else { submitDataWithTimeout(metrics); } } RawMetricsSource(int internalCacheInterval, ExternalMetricsSink rawMetricsSink); @Override void publishTimelineMetrics(Collection<TimelineMetrics> metrics); }
RawMetricsSource implements InternalMetricsSource { @Override public void publishTimelineMetrics(Collection<TimelineMetrics> metrics) { if (rawMetricsSink.getFlushSeconds() > internalCacheInterval) { cache.putAll(metrics); } else { submitDataWithTimeout(metrics); } } RawMetricsSource(int internalCacheInterval, ExternalMetricsSink rawMetricsSink); @Override void publishTimelineMetrics(Collection<TimelineMetrics> metrics);  }