public static Metric metric(String name) {
        return MetricsImpl.metric(name, Unit.COUNT);
    }
@Test
    public void metricsAreJobIndependent() {
        pipeline.readFrom(TestSources.itemStream(1_000))
                .withIngestionTimestamps()
                .filter(l -> l.sequence() < 4000)
                .map(t -> {
                    if (t.sequence() % 4 == 0) {
                        Metrics.metric("total").increment();
                    }
                    return t;
                })
                .writeTo(Sinks.noop());

        Pipeline pipeline2 = Pipeline.create();
        pipeline2.readFrom(TestSources.itemStream(1_000))
                .withIngestionTimestamps()
                .filter(l -> l.sequence() < 4000)
                .map(t -> {
                    if (t.sequence() % 4 != 0) {
                        Metrics.metric("total").increment();
                    }
                    return t;
                })
                .writeTo(Sinks.noop());

        Job job = instance.getJet().newJob(pipeline, JOB_CONFIG_WITH_METRICS);
        Job job2 = instance.getJet().newJob(pipeline2, JOB_CONFIG_WITH_METRICS);
        JobMetricsChecker checker1 = new JobMetricsChecker(job);
        assertTrueEventually(() -> checker1.assertSummedMetricValue("total", 1000));
        JobMetricsChecker checker2 = new JobMetricsChecker(job2);
        assertTrueEventually(() -> checker2.assertSummedMetricValue("total", 3000));
    }