class 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)) {
            // total weight is zero
            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();
    }

    public  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);

}