@Test public void runShouldDoNothingWhenReportConversionFails() { PortBindings portBindings = new PortBindings(Collections.emptyMap(), Collections.emptyMap()); Configuration conf = new Configuration(); Map<String, String> parameters = Collections.singletonMap("reportsDirPath", "/path/to/report"); MetricPusherCreator metricPusherCreator = mock(MetricPusherCreator.class); when(metricPusherCreatorProducer.create(parameters)).thenReturn(Optional.of(metricPusherCreator)); MetricPusher metricPusher = mock(MetricPusher.class); when(metricPusherProducer.create(metricPusherCreator, parameters)).thenReturn(Optional.of(metricPusher)); FileSystem fs = mock(FileSystem.class); when(fileSystemProducer.create(conf)).thenReturn(Optional.of(fs)); when(reportLocationsFinder.find(parameters)).thenReturn(Optional.of(Collections.singletonList("/path/to/report"))); when(labeledMetricConfByPatternProducer.create(parameters)).thenReturn(Optional.of(Collections.emptyMap())); List<ReportEntry> reportEntries = Collections.singletonList(mock(ReportEntry.class)); when(reportEntryReader.read(fs, new Path("/path/to/report"))).thenReturn(Optional.of(reportEntries)); when(reportEntryConverter.convert(anyListOf(ReportEntry.class), any(String.class), anyMapOf(String.class, LabeledMetricConf.class))) .thenReturn(Optional.empty()); pushMetricsProcess.run(portBindings, conf, parameters); verify(metricPusherCreatorProducer, times(1)).create(parameters); verify(metricPusherProducer, times(1)).create(metricPusherCreator, parameters); verify(fileSystemProducer, times(1)).create(conf); verify(reportLocationsFinder, times(1)).find(parameters); verify(labeledMetricConfByPatternProducer, times(1)).create(parameters); verify(reportEntryReader, times(1)).read(fs, new Path("/path/to/report")); verify(reportEntryConverter, times(1)).convert(reportEntries, "/path/to/report", Collections.emptyMap()); verify(metricPusher, never()).pushSafe(any(), any(), any()); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); } }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }  }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters);  }