public static Metric threadSafeMetric(String name) {
        return MetricsImpl.threadSafeMetric(name, Unit.COUNT);
    }
@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.getJet().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();
    }