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 testAggregateWithUpdatedCluster() 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);

    TopicPartition tp1 = new TopicPartition(TOPIC0 + "1", 0);
    Cluster cluster = getCluster(Arrays.asList(TP, tp1));

    List<MetadataResponse.TopicMetadata> topicMetadata = new ArrayList<>(2);
    topicMetadata.add(new MetadataResponse.TopicMetadata(Errors.NONE,
                                                         TOPIC0,
                                                         false,
                                                         Collections.singletonList(new MetadataResponse.PartitionMetadata(
                                                             Errors.NONE, TP, Optional.of(NODE_0.id()),
                                                             Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH),
                                                             nodeIds(), nodeIds(),
                                                             Collections.emptyList()))));
    topicMetadata.add(new MetadataResponse.TopicMetadata(Errors.NONE,
                                                         TOPIC0 + "1",
                                                         false,
                                                         Collections.singletonList(new MetadataResponse.PartitionMetadata(
                                                             Errors.NONE, tp1, Optional.of(NODE_0.id()),
                                                             Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH),
                                                             nodeIds(), nodeIds(),
                                                             Collections.emptyList()))));

    MetadataResponse metadataResponse = KafkaCruiseControlUtils.prepareMetadataResponse(cluster.nodes(),
                                                                                        cluster.clusterResource().clusterId(),
                                                                                        MetadataResponse.NO_CONTROLLER_ID,
                                                                                        topicMetadata);
    metadata.update(KafkaCruiseControlUtils.REQUEST_VERSION_UPDATE, metadataResponse, false, 1);

    Map<PartitionEntity, ValuesAndExtrapolations> aggregateResult =
        metricSampleAggregator.aggregate(cluster, Long.MAX_VALUE, new OperationProgress()).valuesAndExtrapolations();
    // Partition "topic-0" should be valid in all NUM_WINDOW windows and Partition "topic1-0" should not since
    // there is no sample for it.
    assertEquals(1, aggregateResult.size());
    assertEquals(NUM_WINDOWS, aggregateResult.get(PE).windows().size());

    ModelCompletenessRequirements requirements =
        new ModelCompletenessRequirements(1, 0.0, true);
    MetricSampleAggregationResult<String, PartitionEntity> result =
        metricSampleAggregator.aggregate(cluster, -1, Long.MAX_VALUE, requirements, new OperationProgress());
    aggregateResult = result.valuesAndExtrapolations();
    assertNotNull("tp1 should be included because includeAllTopics is set to true",
                  aggregateResult.get(new PartitionEntity(tp1)));
    Map<Integer, Extrapolation> extrapolations = aggregateResult.get(new PartitionEntity(tp1)).extrapolations();
    assertEquals(NUM_WINDOWS, extrapolations.size());

    for (int i = 0; i < NUM_WINDOWS; i++) {
      assertEquals(Extrapolation.NO_VALID_EXTRAPOLATION, extrapolations.get(i));
    }
  }