public double monitoredPercentage(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 completeness.validEntityRatio();
  }
@Test
  public void testMonitoredPercentage() {
    TestContext ctx = setupScenario1();
    KafkaPartitionMetricSampleAggregator aggregator = ctx.aggregator();
    MetadataClient.ClusterAndGeneration clusterAndGeneration = ctx.clusterAndGeneration(0);
    assertEquals(1.0, aggregator.monitoredPercentage(clusterAndGeneration.cluster()), 0.01);

    ctx = setupScenario2();
    aggregator = ctx.aggregator();
    clusterAndGeneration = ctx.clusterAndGeneration(0);
    assertEquals(0.75, aggregator.monitoredPercentage(clusterAndGeneration.cluster()), 0.01);

    ctx = setupScenario3();
    aggregator = ctx.aggregator();
    clusterAndGeneration = ctx.clusterAndGeneration(0);
    assertEquals((double) 4 / 6, aggregator.monitoredPercentage(clusterAndGeneration.cluster()), 0.01);

    ctx = setupScenario4();
    aggregator = ctx.aggregator();
    clusterAndGeneration = ctx.clusterAndGeneration(0);
    assertEquals((double) 4 / 6, aggregator.monitoredPercentage(clusterAndGeneration.cluster()), 0.01);
  }