@Test public void testRawMetricsSourcedAtFlushInterval() throws Exception { InternalSourceProvider internalSourceProvider = new DefaultInternalMetricsSourceProvider(); ExternalMetricsSink rawMetricsSink = createNiceMock(ExternalMetricsSink.class); expect(rawMetricsSink.getFlushSeconds()).andReturn(1); expect(rawMetricsSink.getSinkTimeOutSeconds()).andReturn(1); Capture<Collection<TimelineMetrics>> metricsCapture = new Capture<>(); rawMetricsSink.sinkMetricData(capture(metricsCapture)); expectLastCall(); replay(rawMetricsSink); 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)); verify(rawMetricsSink); }
@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);  }