public static Metric metric(String name) {
        return MetricsImpl.metric(name, Unit.COUNT);
    }
@Test
    public void typicalUsage() {
        pipeline.readFrom(TestSources.items(5L, 4L, 3L, 2L, 1L, 0L))
                .filter(l -> {
                    boolean pass = l % 2 == 0;

                    if (pass) {
                        Metrics.metric("single-flip-flop").decrement();
                        Metrics.metric("multi-flip-flop").decrement(10);
                    } else {
                        Metrics.metric("dropped").increment();
                        Metrics.metric("single-flip-flop").increment();
                        Metrics.metric("multi-flip-flop").increment(10);
                    }
                    Metrics.metric("total").increment();
                    Metrics.metric("last").set(l);

                    return pass;
                })
                .writeTo(Sinks.noop());

        Job job = runPipeline(pipeline.toDag());
        JobMetricsChecker checker = new JobMetricsChecker(job);
        checker.assertSummedMetricValue("dropped", 3);
        checker.assertSummedMetricValue("total", 6);
        checker.assertSummedMetricValue("single-flip-flop", 0);
        checker.assertSummedMetricValue("multi-flip-flop", 0);
    }