class AmbariMetricsServiceWithStormQuerier extends AbstractTimeSeriesQuerier {

    @VisibleForTesting
    Map<Long, Double> aggregateWithApplyingWeightedAverage(Map<Long, List<Pair<String, Double>>> keyMetric,
                                                                   Map<Long, List<Pair<String, Double>>> weightMetric) {
        Map<Long, Double> ret = new HashMap<>();
        for (Map.Entry<Long, List<Pair<String, Double>>> keyMetricEntry : keyMetric.entrySet()) {
            long timestamp = keyMetricEntry.getKey();
            List<Pair<String, Double>> keyStreamToValueList = keyMetricEntry.getValue();
            List<Pair<String, Double>> weightStreamToValueList = weightMetric.get(timestamp);

            if (weightStreamToValueList == null || weightStreamToValueList.isEmpty()) {
                // weight information not found
                ret.put(timestamp, 0.0d);
                continue;
            }

            Double totalWeight = weightStreamToValueList.stream().mapToDouble(p -> p.getRight()).sum();
            if (DoubleUtils.equalsToZero(totalWeight)) {
                // total weight is zero
                ret.put(timestamp, 0.0d);
                continue;
            }

            double weightedSum = keyStreamToValueList.stream().map(pair -> {
                String stream = pair.getLeft();
                Double value = pair.getRight();
                Double weightForStream = weightStreamToValueList.stream()
                        .filter(p -> p.getLeft().equals(stream)).findAny().map(op -> op.getRight()).orElse(0.0);
                Double weight = weightForStream / totalWeight;
                return value * weight;
            }).mapToDouble(d -> d).sum();

            ret.put(timestamp, weightedSum);
        }

        return ret;
    }

    public  AmbariMetricsServiceWithStormQuerier();

}