public double currentConsumptionRate() {
    return aggregateStat(ConsumerCollector.CONSUMER_MESSAGES_PER_SEC, false);
  }
@Test
  public void shouldAggregateStatsAcrossAllConsumers() {
    final MetricCollectors metricCollectors = new MetricCollectors();

    final ConsumerCollector collector1 = new ConsumerCollector();
    collector1.configure(
        ImmutableMap.of(
            ConsumerConfig.CLIENT_ID_CONFIG, "client1",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    final ConsumerCollector collector2 = new ConsumerCollector();
    collector2.configure(
        ImmutableMap.of(
            ConsumerConfig.CLIENT_ID_CONFIG, "client2",
            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);

    // 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(10, Math.floor(metricCollectors.currentConsumptionRate()), 0);
  }