public MetricSampleAggregationResult<String, PartitionEntity> aggregate(Cluster cluster,
                                                                          long now,
                                                                          OperationProgress operationProgress)
      throws NotEnoughValidWindowsException {
    ModelCompletenessRequirements requirements = new ModelCompletenessRequirements(1, 0.0, false);
    return aggregate(cluster, -1L, now, requirements, operationProgress);
  }
@Test
  public void testAggregate() throws NotEnoughValidWindowsException {
    KafkaCruiseControlConfig config = new KafkaCruiseControlConfig(getLoadMonitorProperties());
    Metadata metadata = getMetadata(Collections.singleton(TP));
    KafkaPartitionMetricSampleAggregator
        metricSampleAggregator = new KafkaPartitionMetricSampleAggregator(config, metadata);

    populateSampleAggregator(NUM_WINDOWS + 1, MIN_SAMPLES_PER_WINDOW, metricSampleAggregator);

    MetricSampleAggregationResult<String, PartitionEntity> result =
        metricSampleAggregator.aggregate(metadata.fetch(), Long.MAX_VALUE, new OperationProgress());
    Map<PartitionEntity, ValuesAndExtrapolations> valuesAndExtrapolations = result.valuesAndExtrapolations();

    assertEquals("The windows should only have one partition", 1, valuesAndExtrapolations.size());
    ValuesAndExtrapolations partitionValuesAndExtrapolations = valuesAndExtrapolations.get(PE);
    assertNotNull(partitionValuesAndExtrapolations);
    assertEquals(NUM_WINDOWS, partitionValuesAndExtrapolations.metricValues().length());
    for (int i = 0; i < NUM_WINDOWS; i++) {
      assertEquals((NUM_WINDOWS - i) * WINDOW_MS, result.valuesAndExtrapolations().get(PE).window(i));
      for (Resource resource : Resource.cachedValues()) {
        Collection<Short> metricIds = KafkaMetricDef.resourceToMetricIds(resource);
        double expectedValue = (resource == Resource.DISK ? (NUM_WINDOWS - 1 - i) * 10 + MIN_SAMPLES_PER_WINDOW - 1
                                                          : (NUM_WINDOWS - 1 - i) * 10 + (MIN_SAMPLES_PER_WINDOW - 1) / 2.0)
                               / (resource == Resource.CPU ? UNIT_INTERVAL_TO_PERCENTAGE : 1.0) * metricIds.size();
        assertEquals("The utilization for " + resource + " should be " + expectedValue,
                     expectedValue, partitionValuesAndExtrapolations.metricValues().valuesForGroup(resource.name(),
                                                                                                   KafkaMetricDef.commonMetricDef(),
                                                                                                   true).get(i), 0.01);
      }
    }

    // Verify the metric completeness checker state
    MetadataClient.ClusterAndGeneration clusterAndGeneration =
        new MetadataClient.ClusterAndGeneration(metadata.fetch(), 1);
    assertEquals(NUM_WINDOWS, metricSampleAggregator.validWindows(clusterAndGeneration.cluster(), 1.0).size());
    Map<Long, Float> monitoredPercentages = metricSampleAggregator.validPartitionRatioByWindows(clusterAndGeneration.cluster());
    for (double percentage : monitoredPercentages.values()) {
      assertEquals(1.0, percentage, 0.0);
    }
    assertEquals(NUM_WINDOWS, metricSampleAggregator.availableWindows().size());
  }