ComponentMetricSummary { @VisibleForTesting static double calculateWeightedAverage(Map<Long, Double> keyMetrics, Map<Long, Double> weightMetrics) { Map<Long, Double> filteredKeyMetrics = keyMetrics.entrySet().stream() .filter(entry -> entry.getValue() != null && DoubleUtils.notEqualsToZero(entry.getValue())) .collect(toMap(e -> e.getKey(), e -> e.getValue())); Map<Long, Double> filteredWeightMetrics = weightMetrics.entrySet().stream() .filter(entry -> filteredKeyMetrics.containsKey(entry.getKey())) .collect(toMap(e -> e.getKey(), e -> e.getValue())); Double sumInputRecords = filteredWeightMetrics.values().stream().mapToDouble(d -> d).sum(); if (DoubleUtils.equalsToZero(sumInputRecords)) { return 0.0d; } return filteredKeyMetrics.entrySet().stream() .map(entry -> { Long timestamp = entry.getKey(); double weight = filteredWeightMetrics.getOrDefault(timestamp, 0.0) / sumInputRecords; return entry.getValue() * weight; }).mapToDouble(d -> d).sum(); } ComponentMetricSummary(Long emitted, Long acked, Long failed, Double latency, Double completeLatency,
                                  Double processTime, Double executeTime,
                                  Long prevEmitted, Long prevAcked, Long prevFailed, Double prevLatency,
                                  Double prevCompleteLatency, Double prevProcessTime, Double prevExecuteTime); Long getEmitted(); Long getAcked(); Long getFailed(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getLatency(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getCompleteLatency(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getProcessTime(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getExecuteTime(); Long getPrevEmitted(); Long getPrevAcked(); Long getPrevFailed(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getPrevLatency(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getPrevCompleteLatency(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getPrevProcessTime(); @JsonInclude(JsonInclude.Include.NON_NULL) Double getPrevExecuteTime(); static ComponentMetricSummary convertTopologyMetric(TopologyTimeSeriesMetrics.TimeSeriesComponentMetric metrics,
                                                               TopologyTimeSeriesMetrics.TimeSeriesComponentMetric prevMetrics); static ComponentMetricSummary convertSourceMetric(TopologyTimeSeriesMetrics.TimeSeriesComponentMetric metrics,
                                                             TopologyTimeSeriesMetrics.TimeSeriesComponentMetric prevMetrics); static ComponentMetricSummary convertNonSourceMetric(TopologyTimeSeriesMetrics.TimeSeriesComponentMetric metrics,
                                                                TopologyTimeSeriesMetrics.TimeSeriesComponentMetric prevMetrics);  }
@Test public void testCalculateWeightedAverage() { Map<Long, Double> keyMetrics = new HashMap<>(); keyMetrics.put(1L, 10.0); keyMetrics.put(2L, 20.0); Map<Long, Double> weightMetrics = new HashMap<>(); weightMetrics.put(1L, 10.0); weightMetrics.put(2L, 5.0); double actual = ComponentMetricSummary.calculateWeightedAverage(keyMetrics, weightMetrics); double expected = 13.333333333d; Assert.assertEquals(expected, actual, DoubleUtils.EPSILON); }
@Test public void testCalculateWeightedAverageLacksWeightInformation() { Map<Long, Double> keyMetrics = new HashMap<>(); keyMetrics.put(1L, 10.0); keyMetrics.put(2L, 20.0); double actual = ComponentMetricSummary.calculateWeightedAverage(keyMetrics, Collections.emptyMap()); Assert.assertEquals(0.0, actual, DoubleUtils.EPSILON); Map<Long, Double> weightMetrics = new HashMap<>(); weightMetrics.put(2L, 5.0); actual = ComponentMetricSummary.calculateWeightedAverage(keyMetrics, weightMetrics); Assert.assertEquals(20.0, actual, DoubleUtils.EPSILON); }