@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, PARTITION, NODE_0, Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH), Arrays.asList(nodes()), Arrays.asList(nodes()), Collections.emptyList())))); topicMetadata.add(new MetadataResponse.TopicMetadata(Errors.NONE, TOPIC0 + "1", false, Collections.singletonList(new MetadataResponse.PartitionMetadata( Errors.NONE, 0, NODE_0, Optional.of(RecordBatch.NO_PARTITION_LEADER_EPOCH), Arrays.asList(nodes()), Arrays.asList(nodes()), Collections.emptyList())))); MetadataResponse metadataResponse = KafkaCruiseControlUtils.prepareMetadataResponse(cluster.nodes(), cluster.clusterResource().clusterId(), MetadataResponse.NO_CONTROLLER_ID, topicMetadata); metadata.update(KafkaCruiseControlUtils.REQUEST_VERSION_UPDATE, metadataResponse, 1); Map<PartitionEntity, ValuesAndExtrapolations> aggregateResult = metricSampleAggregator.aggregate(cluster, Long.MAX_VALUE, new OperationProgress()).valuesAndExtrapolations(); 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)); } }
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);  }