Metrics { private Metrics() { } private  Metrics(); static Metric metric(String name); static Metric metric(String name, Unit unit); static Metric threadSafeMetric(String name); static Metric threadSafeMetric(String name, Unit unit);  }
@Test public void test_sourceSinkTag() { DAG dag = new DAG(); Vertex src = dag.newVertex("src", () -> new NoOutputSourceP()); Vertex mid = dag.newVertex("mid", Processors.mapP(identity())); Vertex sink = dag.newVertex("sink", Processors.noopP()); dag.edge(Edge.between(src, mid)); dag.edge(Edge.between(mid, sink)); Job job = instance().newJob(dag, new JobConfig() .setProcessingGuarantee(EXACTLY_ONCE) .setSnapshotIntervalMillis(100)); assertJobStatusEventually(job, RUNNING); JobMetrics[] metrics = {null}; assertTrueEventually(() -> assertNotEquals(0, (metrics[0] = job.getMetrics()).metrics().size())); assertSourceSinkTags(metrics[0], "src", true, true, false); assertSourceSinkTags(metrics[0], "mid", true, false, false); assertSourceSinkTags(metrics[0], "sink", true, false, true); waitForFirstSnapshot(new JobRepository(instance()), job.getId(), 10, true); job.restart(); assertJobStatusEventually(job, RUNNING); assertTrueEventually(() -> assertNotEquals(0, (metrics[0] = job.getMetrics()).metrics().size())); assertSourceSinkTags(metrics[0], "src", false, true, false); assertSourceSinkTags(metrics[0], "mid", false, false, false); assertSourceSinkTags(metrics[0], "sink", false, false, true); }