public Collection<Double> currentConsumptionRateByQuery() {
    return collectorMap.values()
        .stream()
        .filter(collector -> collector.getGroupId() != null)
        .collect(
            Collectors.groupingBy(
              MetricCollector::getGroupId,
              Collectors.summingDouble(
                  m -> m.aggregateStat(ConsumerCollector.CONSUMER_MESSAGES_PER_SEC, false)
              )
          )
        )
        .values();
  }
@Test
  public void shouldNotIncludeRestoreConsumersWhenComputingPerQueryStats() {
    final MetricCollectors metricCollectors = new MetricCollectors();
    final ConsumerCollector collector1 = new ConsumerCollector();
    collector1.configure(
        ImmutableMap.of(
            ConsumerConfig.GROUP_ID_CONFIG, "group1",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    final ConsumerCollector collector2 = new ConsumerCollector();
    collector2.configure(
        ImmutableMap.of(
            ConsumerConfig.GROUP_ID_CONFIG, "group1",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    final ConsumerCollector collector3 = new ConsumerCollector();
    collector3.configure(
        ImmutableMap.of(
            ConsumerConfig.GROUP_ID_CONFIG, "group2",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    // The restore consumer doesn't have a group id, and hence we should not count it as part of
    // the overall query stats.
    final ConsumerCollector collector4 = new ConsumerCollector();
    collector4.configure(
        ImmutableMap.of(
            ConsumerConfig.CLIENT_ID_CONFIG, "restore-consumer-client",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );


    final Map<TopicPartition, List<ConsumerRecord<Object, Object>>> records = new HashMap<>();
    final List<ConsumerRecord<Object, Object>> recordList = new ArrayList<>();
    for (int i = 0; i < 500; i++) {
      recordList.add(
          new ConsumerRecord<>(
              TEST_TOPIC,
              1,
              1,
              1L,
              TimestampType
                  .CREATE_TIME,
              10,
              10,
              "key",
              "1234567890",
              new RecordHeaders(),
              Optional.empty()
          )
      );
    }
    records.put(new TopicPartition(TEST_TOPIC, 1), recordList);
    final ConsumerRecords<Object, Object> consumerRecords = new ConsumerRecords<>(records);
    collector1.onConsume(consumerRecords);
    collector2.onConsume(consumerRecords);
    collector3.onConsume(consumerRecords);
    collector4.onConsume(consumerRecords);

    final List<Double> consumptionByQuery = new ArrayList<>(
        metricCollectors.currentConsumptionRateByQuery());
    consumptionByQuery.sort(Comparator.naturalOrder());

    // Each query will have a unique consumer group id. In this case we have two queries and 3
    // consumers. So we should expect two results from the currentConsumptionRateByQuery call.
    assertEquals(2, consumptionByQuery.size());

    // Same as the above test, the kafka `Rate` measurable stat reports the rate as a tenth
    // of what it should be because all the samples haven't been filled out yet.
    assertEquals(5.0, Math.floor(consumptionByQuery.get(0)), 0.1);
    assertEquals(10.0, Math.floor(consumptionByQuery.get(1)), 0.1);
  }