YarnJobValidationTool { public int validateContainerCount(ApplicationAttemptId attemptId) throws Exception { int runningContainerCount = 0; for (ContainerReport containerReport : this.client.getContainers(attemptId)) { if (containerReport.getContainerState() == ContainerState.RUNNING) { ++runningContainerCount; } } int containerExpected = this.config.getContainerCount() + 1; if (runningContainerCount == containerExpected) { log.info("Container count matches. " + runningContainerCount + " containers are running."); return runningContainerCount; } else { throw new SamzaException("Container count does not match. " + runningContainerCount + " containers are running, while " + containerExpected + " is expected."); } } 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 testValidateContainerCount() throws Exception { List<ContainerReport> containerReports = new ArrayList<>(); for (int i = 0; i <= containerCount; i++) { ContainerReport report = mock(ContainerReport.class); when(report.getContainerState()).thenReturn(ContainerState.RUNNING); containerReports.add(report); } when(client.getContainers(attemptId)).thenReturn(containerReports); assertTrue(tool.validateContainerCount(attemptId) == (containerCount + 1)); containerReports.remove(0); exception.expect(SamzaException.class); tool.validateContainerCount(attemptId); }