public static Metric metric(String name) {
        return MetricsImpl.metric(name, Unit.COUNT);
    }
@Test
    public void nonCooperativeProcessor() {
        DAG dag = new DAG();

        Vertex source = dag.newVertex("source", TestProcessors.ListSource.supplier(asList(1L, 2L, 3L)));
        Vertex map = dag.newVertex("map", new NonCoopTransformPSupplier((FunctionEx<Long, Long>) l -> {
            Metrics.metric("mapped").increment();
            return l * 10L;
        }));
        Vertex sink = dag.newVertex("sink", writeListP("results"));

        dag.edge(between(source, map)).edge(between(map, sink));

        Job job = runPipeline(dag);
        JobMetricsChecker checker = new JobMetricsChecker(job);
        checker.assertSummedMetricValue("mapped", 3L);
        assertEquals(
                new HashSet<>(Arrays.asList(10L, 20L, 30L)),
                new HashSet<>(instance.getList("results"))
        );
    }