### Question:
MetricCollectors { public static double totalMessageConsumption() { return collectorMap.values().stream() .mapToDouble(MetricCollector::totalMessageConsumption) .sum(); }  static void initialize(); static void cleanUp(); static String getStatsFor(final String topic, boolean isError); static void recordError(String topic); static Collection<Double> currentConsumptionRateByQuery(); static double currentProductionRate(); static double currentConsumptionRate(); static double totalMessageConsumption(); static double totalBytesConsumption(); static double currentErrorRate(); static Metrics getMetrics(); static Time getTime();  }### Answer:
@Test public void shouldAggregateTotalMessageConsumptionAcrossAllConsumers() throws Exception { ConsumerCollector collector1 = new ConsumerCollector(); collector1.configure(ImmutableMap.of(ConsumerConfig.CLIENT_ID_CONFIG, "client1")); ConsumerCollector collector2 = new ConsumerCollector(); collector2.configure(ImmutableMap.of(ConsumerConfig.CLIENT_ID_CONFIG, "client2")); Map<TopicPartition, List<ConsumerRecord<Object, Object>>> records = new HashMap<>(); List<ConsumerRecord<Object, Object>> recordList = new ArrayList<>(); for (int i = 0; i < 10; i++) { recordList.add(new ConsumerRecord<>(TEST_TOPIC, 1, 1, 1l, TimestampType .CREATE_TIME, 1l, 10, 10,"key", "1234567890")); } records.put(new TopicPartition(TEST_TOPIC, 1), recordList); ConsumerRecords<Object, Object> consumerRecords = new ConsumerRecords<>(records); collector1.onConsume(consumerRecords); collector2.onConsume(consumerRecords); assertEquals(20, MetricCollectors.totalMessageConsumption(), 0); }