public double currentProductionRate() {
    return aggregateStat(ProducerCollector.PRODUCER_MESSAGES_PER_SEC, false);
  }
@Test
  public void shouldAggregateStatsAcrossAllProducers() {
    final MetricCollectors metricCollectors = new MetricCollectors();

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

    final ProducerCollector collector2 = new ProducerCollector();
    collector2.configure(
        ImmutableMap.of(
            ProducerConfig.CLIENT_ID_CONFIG, "client2",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    for (int i = 0; i < 500; i++) {
      collector1.onSend(new ProducerRecord<>(TEST_TOPIC, "key", Integer.toString(i)));
      collector2.onSend(new ProducerRecord<>(TEST_TOPIC + "_" + i, "key",
                                             Integer.toString(i * 100)));
    }

    // The Kafka metrics in MetricCollectors is configured so that sampled stats (like the Rate
    // measurable stat) have a 100 samples, each with a duration of 1 second. In this test we
    // record a 1000 events, but only in a single sample since they all belong to the same second.
    // So 99 samples are empty. Hence the rate is computed as a tenth of what it should be. This
    // won't be a problem for a longer running program.
    assertEquals(10, Math.floor(metricCollectors.currentProductionRate()), 0);
  }