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 testFallbackToAvgAdjacent() throws NotEnoughValidWindowsException {
    KafkaCruiseControlConfig config = new KafkaCruiseControlConfig(getLoadMonitorProperties());
    TopicPartition anotherTopicPartition = new TopicPartition("AnotherTopic", 1);
    PartitionEntity anotherPartitionEntity = new PartitionEntity(anotherTopicPartition);
    Metadata metadata = getMetadata(Arrays.asList(TP, anotherTopicPartition));
    KafkaPartitionMetricSampleAggregator
        metricSampleAggregator = new KafkaPartitionMetricSampleAggregator(config, metadata);

    // Only give one sample to the aggregator for previous period.
    populateSampleAggregator(NUM_WINDOWS, MIN_SAMPLES_PER_WINDOW, metricSampleAggregator);
    // Create let (NUM_WINDOWS + 1) have enough samples.
    CruiseControlUnitTestUtils.populateSampleAggregator(1, MIN_SAMPLES_PER_WINDOW, metricSampleAggregator,
                                                        PE, NUM_WINDOWS, WINDOW_MS,
                                                        KafkaMetricDef.commonMetricDef());
    // Let a window exist but not containing samples for partition 0
    CruiseControlUnitTestUtils.populateSampleAggregator(1, MIN_SAMPLES_PER_WINDOW, metricSampleAggregator,
                                                        anotherPartitionEntity, NUM_WINDOWS + 1, WINDOW_MS, KafkaMetricDef
                                                            .commonMetricDef());
    // Let the rest of the window has enough samples.
    CruiseControlUnitTestUtils.populateSampleAggregator(2, MIN_SAMPLES_PER_WINDOW,
                                                        metricSampleAggregator, PE,
                                                        NUM_WINDOWS + 2, WINDOW_MS,
                                                        KafkaMetricDef.commonMetricDef());

    MetricSampleAggregationResult<String, PartitionEntity> result =
        metricSampleAggregator.aggregate(metadata.fetch(), NUM_WINDOWS * WINDOW_MS * 2, new OperationProgress());
    int numWindows = result.valuesAndExtrapolations().get(PE).metricValues().length();
    assertEquals(NUM_WINDOWS, numWindows);
    int numExtrapolations = 0;
    for (Map.Entry<Integer, Extrapolation> entry : result.valuesAndExtrapolations().get(PE).extrapolations().entrySet()) {
      assertEquals(Extrapolation.AVG_ADJACENT, entry.getValue());
      numExtrapolations++;
    }
    assertEquals(1, numExtrapolations);
  }