YarnJobValidationTool { public void validateJmxMetrics() throws Exception { MetricsRegistry metricsRegistry = new MetricsRegistryMap(); CoordinatorStreamStore coordinatorStreamStore = new CoordinatorStreamStore(config, metricsRegistry); coordinatorStreamStore.init(); try { LocalityManager localityManager = new LocalityManager(new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore, SetConfig.TYPE)); validator.init(config); LocalityModel localityModel = localityManager.readLocality(); for (ProcessorLocality processorLocality : localityModel.getProcessorLocalities().values()) { String containerId = processorLocality.id(); String jmxUrl = processorLocality.jmxTunnelingUrl(); if (StringUtils.isNotBlank(jmxUrl)) { log.info("validate container " + containerId + " metrics with JMX: " + jmxUrl); JmxMetricsAccessor jmxMetrics = new JmxMetricsAccessor(jmxUrl); jmxMetrics.connect(); validator.validate(jmxMetrics); jmxMetrics.close(); log.info("validate container " + containerId + " successfully"); } } validator.complete(); } finally { coordinatorStreamStore.close(); } } YarnJobValidationTool(JobConfig config, YarnClient client, MetricsValidator validator); void run(); ApplicationId validateAppId(); ApplicationAttemptId validateRunningAttemptId(ApplicationId appId); int validateContainerCount(ApplicationAttemptId attemptId); void validateJmxMetrics(); static void main(String[] args);  }
@Test public void testValidateJmxMetrics() throws MetricsValidationFailureException { JmxMetricsAccessor jmxMetricsAccessor = mock(JmxMetricsAccessor.class); Map<String, Long> values = new HashMap<>(); values.put("samza-container-0", 100L); when(jmxMetricsAccessor.getCounterValues(SamzaContainerMetrics.class.getName(), "commit-calls")).thenReturn(values); validator.validate(jmxMetricsAccessor); values.put("samza-container-0", -1L); exception.expect(MetricsValidationFailureException.class); validator.validate(jmxMetricsAccessor); }