InstrumentedJobListener implements JobListener { @Override public void onJobStarted(String jobName, Map<String, Object> parameters, Consumer<Consumer<JobResult>> finishEventSource) { JobMetrics metric = getOrCreateMetrics(jobName); metric.getActiveCounter().inc(); Timer.Context requestTimerContext = metric.getTimer().time(); LOGGER.info("started job: '{}'", jobName); finishEventSource.accept(result -> { metric.getActiveCounter().dec(); metric.getCompletedCounter().inc(); long timeNanos = requestTimerContext.stop(); LOGGER.info("finished job '{}' in {} ms", jobName, timeNanos / 1000000); switch (result.getOutcome()) { case SUCCESS: { metric.getSuccessCounter().inc(); break; } case FAILURE: { metric.getFailureCounter().inc(); break; } } }); } InstrumentedJobListener(MetricRegistry metricRegistry); @Override void onJobStarted(String jobName, Map<String, Object> parameters, Consumer<Consumer<JobResult>> finishEventSource);  }
@Test public void testJobsInstrumentation_ActiveCount_SuccessAndFailureResults() { InstrumentedJobListener listener = new InstrumentedJobListener(metricRegistry); FinishEventSource f1 = new FinishEventSource(); listener.onJobStarted("j1", Collections.emptyMap(), f1); assertHasMetrics("j1", metricRegistry, 1, 0, 0, 0); FinishEventSource f2 = new FinishEventSource(); listener.onJobStarted("j1", Collections.emptyMap(), f2); assertHasMetrics("j1", metricRegistry, 2, 0, 0, 0); f1.finish(JobResult.success(null)); assertHasMetrics("j1", metricRegistry, 1, 1, 1, 0); f2.finish(JobResult.failure(null)); assertHasMetrics("j1", metricRegistry, 0, 2, 1, 1); }
@Test public void testJobsInstrumentation_UnknownResult() { InstrumentedJobListener listener = new InstrumentedJobListener(metricRegistry); FinishEventSource f1 = new FinishEventSource(); listener.onJobStarted("j1", Collections.emptyMap(), f1); assertHasMetrics("j1", metricRegistry, 1, 0, 0, 0); f1.finish(JobResult.unknown(null)); assertHasMetrics("j1", metricRegistry, 0, 1, 0, 0); }
@Test public void testJobsInstrumentation_SuccessResult() { FinishEventSource finishEventSource = new FinishEventSource(); InstrumentedJobListener listener = new InstrumentedJobListener(metricRegistry); listener.onJobStarted("j1", Collections.emptyMap(), finishEventSource); finishEventSource.finish(JobResult.unknown(null)); assertHasMetrics("j1", metricRegistry, 0, 1, 0, 0); }