Metrics { public static Metric threadSafeMetric(String name) { return MetricsImpl.threadSafeMetric(name, Unit.COUNT); } private  Metrics(); static Metric metric(String name); static Metric metric(String name, Unit unit); static Metric threadSafeMetric(String name); static Metric threadSafeMetric(String name, Unit unit);  }
@Test public void usingServiceAsync() { int inputSize = 100_000; Integer[] inputs = new Integer[inputSize]; Arrays.setAll(inputs, i -> i); pipeline.readFrom(TestSources.items(inputs)) .addTimestamps(i -> i, 0L) .mapUsingServiceAsync( nonSharedService(pctx -> 0L), (ctx, l) -> { Metric dropped = Metrics.threadSafeMetric("dropped"); Metric total = Metrics.threadSafeMetric("total"); return CompletableFuture.supplyAsync( () -> { boolean pass = l % 2L == ctx; if (!pass) { dropped.increment(); } total.increment(); return l; } ); } ) .writeTo(Sinks.noop()); Job job = instance().newJob(pipeline, JOB_CONFIG_WITH_METRICS); assertTrueEventually(() -> assertEquals(inputSize, job.getMetrics().get("total").stream().mapToLong(Measurement::value).sum())); assertTrueEventually(() -> assertEquals(inputSize / 2, job.getMetrics().get("dropped").stream().mapToLong(Measurement::value).sum())); job.join(); }