@Test public void testAggregate() 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); MetricSampleAggregationResult<String, PartitionEntity> result = metricSampleAggregator.aggregate(metadata.fetch(), Long.MAX_VALUE, new OperationProgress()); Map<PartitionEntity, ValuesAndExtrapolations> valuesAndExtrapolations = result.valuesAndExtrapolations(); assertEquals("The windows should only have one partition", 1, valuesAndExtrapolations.size()); ValuesAndExtrapolations partitionValuesAndExtrapolations = valuesAndExtrapolations.get(PE); assertNotNull(partitionValuesAndExtrapolations); assertEquals(NUM_WINDOWS, partitionValuesAndExtrapolations.metricValues().length()); for (int i = 0; i < NUM_WINDOWS; i++) { assertEquals((NUM_WINDOWS - i) * WINDOW_MS, result.valuesAndExtrapolations().get(PE).window(i)); for (Resource resource : Resource.cachedValues()) { Collection<Short> metricIds = KafkaMetricDef.resourceToMetricIds(resource); double expectedValue = (resource == Resource.DISK ? (NUM_WINDOWS - 1 - i) * 10 + MIN_SAMPLES_PER_WINDOW - 1 : (NUM_WINDOWS - 1 - i) * 10 + (MIN_SAMPLES_PER_WINDOW - 1) / 2.0) / (resource == Resource.CPU ? UNIT_INTERVAL_TO_PERCENTAGE : 1.0) * metricIds.size(); assertEquals("The utilization for " + resource + " should be " + expectedValue, expectedValue, partitionValuesAndExtrapolations.metricValues().valuesForGroup(resource.name(), KafkaMetricDef.commonMetricDef(), true).get(i), 0.01); } } MetadataClient.ClusterAndGeneration clusterAndGeneration = new MetadataClient.ClusterAndGeneration(metadata.fetch(), 1); assertEquals(NUM_WINDOWS, metricSampleAggregator.validWindows(clusterAndGeneration.cluster(), 1.0).size()); Map<Long, Float> monitoredPercentages = metricSampleAggregator.validPartitionRatioByWindows(clusterAndGeneration.cluster()); for (double percentage : monitoredPercentages.values()) { assertEquals(1.0, percentage, 0.0); } assertEquals(NUM_WINDOWS, metricSampleAggregator.availableWindows().size()); }
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); }
KafkaPartitionMetricSampleAggregator extends MetricSampleAggregator<String, PartitionEntity> { 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); } }
KafkaPartitionMetricSampleAggregator extends MetricSampleAggregator<String, PartitionEntity> { 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); } KafkaPartitionMetricSampleAggregator(KafkaCruiseControlConfig config, Metadata metadata); }
KafkaPartitionMetricSampleAggregator extends MetricSampleAggregator<String, PartitionEntity> { 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); } KafkaPartitionMetricSampleAggregator(KafkaCruiseControlConfig config, Metadata metadata); boolean addSample(PartitionMetricSample sample); boolean addSample(PartitionMetricSample sample, boolean leaderValidation); MetricSampleAggregationResult<String, PartitionEntity> aggregate(Cluster cluster,
                                                                          long now,
                                                                          OperationProgress operationProgress); MetricSampleAggregationResult<String, PartitionEntity> aggregate(Cluster cluster,
                                                                          long from,
                                                                          long to,
                                                                          ModelCompletenessRequirements requirements,
                                                                          OperationProgress operationProgress); MetricSampleCompleteness<String, PartitionEntity> completeness(Cluster cluster,
                                                                        long from,
                                                                        long to,
                                                                        ModelCompletenessRequirements requirements); SortedSet<Long> validWindows(Cluster cluster, double minMonitoredPartitionsPercentage); double monitoredPercentage(Cluster cluster); SortedMap<Long, Float> validPartitionRatioByWindows(Cluster cluster); }
KafkaPartitionMetricSampleAggregator extends MetricSampleAggregator<String, PartitionEntity> { 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); } KafkaPartitionMetricSampleAggregator(KafkaCruiseControlConfig config, Metadata metadata); boolean addSample(PartitionMetricSample sample); boolean addSample(PartitionMetricSample sample, boolean leaderValidation); MetricSampleAggregationResult<String, PartitionEntity> aggregate(Cluster cluster,
                                                                          long now,
                                                                          OperationProgress operationProgress); MetricSampleAggregationResult<String, PartitionEntity> aggregate(Cluster cluster,
                                                                          long from,
                                                                          long to,
                                                                          ModelCompletenessRequirements requirements,
                                                                          OperationProgress operationProgress); MetricSampleCompleteness<String, PartitionEntity> completeness(Cluster cluster,
                                                                        long from,
                                                                        long to,
                                                                        ModelCompletenessRequirements requirements); SortedSet<Long> validWindows(Cluster cluster, double minMonitoredPartitionsPercentage); double monitoredPercentage(Cluster cluster); SortedMap<Long, Float> validPartitionRatioByWindows(Cluster cluster);  }