public static Metric metric(String name) {
        return MetricsImpl.metric(name, Unit.COUNT);
    }
@Test
    public void metricInNotFusedStages() {
        int inputSize = 100_000;
        Integer[] inputs = new Integer[inputSize];
        Arrays.setAll(inputs, i -> i);

        pipeline.readFrom(TestSources.items(inputs))
                .filter(l -> {
                    Metrics.metric("onlyInFilter").increment();
                    Metrics.metric("inBoth").increment();
                    return true;
                })
                .groupingKey(t -> t)
                .aggregate(counting())
                .map(t -> {
                    Metrics.metric("onlyInMap").increment();
                    Metrics.metric("inBoth").increment();
                    return t;
                })
                .writeTo(Sinks.noop());

        Job job = runPipeline(pipeline.toDag());
        JobMetricsChecker checker = new JobMetricsChecker(job);
        checker.assertSummedMetricValue("onlyInFilter", inputSize);
        checker.assertSummedMetricValue("onlyInMap", inputSize);
        checker.assertSummedMetricValue("inBoth", 2 * inputSize);
    }