public SortedSet<Long> validWindows(Cluster cluster, double minMonitoredPartitionsPercentage) {
    AggregationOptions<String, PartitionEntity> options = new AggregationOptions<>(minMonitoredPartitionsPercentage,
                                                                                   0.0,
                                                                                   1,
                                                                                   _maxAllowedExtrapolationsPerPartition,
                                                                                   allPartitions(cluster),
                                                                                   AggregationOptions.Granularity.ENTITY,
                                                                                   true);
    MetricSampleCompleteness<String, PartitionEntity> completeness = completeness(-1, Long.MAX_VALUE, options);
    return windowIndicesToWindows(completeness.validWindowIndices(), _windowMs);
  }
@Test
  public void testValidWindows() {
    TestContext ctx = setupScenario1();
    KafkaPartitionMetricSampleAggregator aggregator = ctx.aggregator();
    MetadataClient.ClusterAndGeneration clusterAndGeneration = ctx.clusterAndGeneration(0);
    SortedSet<Long> validWindows = aggregator.validWindows(clusterAndGeneration.cluster(), 1.0);
    assertEquals(NUM_WINDOWS, validWindows.size());
    assertValidWindows(validWindows, NUM_WINDOWS, Collections.emptySet());
  }