public SortedMap<Long, Float> validPartitionRatioByWindows(Cluster cluster) {
    AggregationOptions<String, PartitionEntity> options = new AggregationOptions<>(0.0,
                                                                                   0.0,
                                                                                   1,
                                                                                   _maxAllowedExtrapolationsPerPartition,
                                                                                   allPartitions(cluster),
                                                                                   AggregationOptions.Granularity.ENTITY,
                                                                                   true);
    MetricSampleCompleteness<String, PartitionEntity> completeness = completeness(-1, Long.MAX_VALUE, options);
    return windowIndicesToWindows(completeness.validEntityRatioWithGroupGranularityByWindowIndex(), _windowMs);
  }
@Test
  public void testMonitoredPercentagesByWindows() {
    TestContext ctx = setupScenario1();
    KafkaPartitionMetricSampleAggregator aggregator = ctx.aggregator();
    MetadataClient.ClusterAndGeneration clusterAndGeneration = ctx.clusterAndGeneration(0);

    Map<Long, Float> percentages = aggregator.validPartitionRatioByWindows(clusterAndGeneration.cluster());
    assertEquals(NUM_WINDOWS, percentages.size());
    for (Map.Entry<Long, Float> entry : percentages.entrySet()) {
      assertEquals(1.0, entry.getValue(), 0.01);
    }

    ctx = setupScenario2();
    aggregator = ctx.aggregator();
    clusterAndGeneration = ctx.clusterAndGeneration(0);
    percentages = aggregator.validPartitionRatioByWindows(clusterAndGeneration.cluster());
    assertEquals(NUM_WINDOWS, percentages.size());
    for (Map.Entry<Long, Float> entry : percentages.entrySet()) {
      long window = entry.getKey();
      if (window == 6000 || window == 7000 || window == 20000) {
        assertEquals(0.5, entry.getValue(), 0.01);
      } else {
        assertEquals(1.0, entry.getValue(), 0.01);
      }
    }

    ctx = setupScenario3();
    aggregator = ctx.aggregator();
    clusterAndGeneration = ctx.clusterAndGeneration(0);
    percentages = aggregator.validPartitionRatioByWindows(clusterAndGeneration.cluster());
    assertEquals(NUM_WINDOWS, percentages.size());
    for (Map.Entry<Long, Float> entry : percentages.entrySet()) {
      long window = entry.getKey();
      if (window == 6000 || window == 7000 || window == 18000 || window == 19000) {
        assertEquals((double) 4 / 6, entry.getValue(), 0.01);
      } else {
        assertEquals(1.0, entry.getValue(), 0.01);
      }
    }

    ctx = setupScenario4();
    aggregator = ctx.aggregator();
    clusterAndGeneration = ctx.clusterAndGeneration(0);
    percentages = aggregator.validPartitionRatioByWindows(clusterAndGeneration.cluster());
    assertEquals(NUM_WINDOWS, percentages.size());
    for (Map.Entry<Long, Float> entry : percentages.entrySet()) {
      long window = entry.getKey();
      if (window == 6000 || window == 7000) {
        assertEquals((double) 2 / 6, entry.getValue(), 0.01);
      } else {
        assertEquals((double) 4 / 6, entry.getValue(), 0.01);
      }
    }
  }