PigCountersReportGenerator implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { String pigCountersJson = parameters.get(PIG_COUNTERS_PROPERTY); PigCounters pigCounters = pigCountersParser.parse(pigCountersJson); List<ReportPigCounterMapping> reportCountersMapping = collectReportCountersMapping(parameters); List<ReportEntry> reportCounters = reportPigCountersResolver.resolveReportCounters(pigCounters, reportCountersMapping); FileSystem fs = FileSystem.get(conf); Path reportPath = portBindings.getOutput().get(REPORT_PORT_OUT_NAME); DataStore.create(reportCounters, new FileSystemPath(fs, reportPath)); }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters);  }
@Test public void run() throws Exception { Path outputDirPath = new Path(tempFolder.getRoot().getPath()); PortBindings portBindings = new PortBindings(ImmutableMap.of(), ImmutableMap.of("report", outputDirPath)); Configuration conf = new Configuration(false); Map<String, String> parameters = ImmutableMap.of( "pigCounters", "counters", "report.group.param1", "pigCounterName1", "report.group.param2", "pigCounterName2"); PigCounters pigCounters = Mockito.mock(PigCounters.class); when(pigCountersParser.parse("counters")).thenReturn(pigCounters); ReportPigCounterMapping counterMapping1 = new ReportPigCounterMapping("counterName1", "jobAlias1", "group.param1"); ReportPigCounterMapping counterMapping2 = new ReportPigCounterMapping("counterName2", "jobAlias2", "group.param2"); when(reportPigCounterMappingParser.parse("group.param1", "pigCounterName1")).thenReturn(counterMapping1); when(reportPigCounterMappingParser.parse("group.param2", "pigCounterName2")).thenReturn(counterMapping2); ReportEntry reportCounter1 = new ReportEntry("group.param1", ReportEntryType.COUNTER, "2"); ReportEntry reportCounter2 = new ReportEntry("group.param2", ReportEntryType.COUNTER, "8"); when(reportPigCountersResolver.resolveReportCounters(pigCounters, Lists.newArrayList(counterMapping1, counterMapping2))) .thenReturn(Lists.newArrayList(reportCounter1, reportCounter2)); pigCountersReportGenerator.run(portBindings, conf, parameters); List<ReportEntry> actualReportCounters = AvroTestUtils.readLocalAvroDataStore(tempFolder.getRoot().getPath()); assertThat(actualReportCounters, containsInAnyOrder(reportCounter1, reportCounter2)); }