### Question:
KsqlEngineMetrics implements Closeable { public void updateMetrics() { recordMessagesConsumed(MetricCollectors.currentConsumptionRate()); recordTotalMessagesConsumed(MetricCollectors.totalMessageConsumption()); recordTotalBytesConsumed(MetricCollectors.totalBytesConsumption()); recordMessagesProduced(MetricCollectors.currentProductionRate()); recordMessageConsumptionByQueryStats(MetricCollectors.currentConsumptionRateByQuery()); recordErrorRate(MetricCollectors.currentErrorRate()); } KsqlEngineMetrics(String metricGroupPrefix, KsqlEngine ksqlEngine); @Override void close(); void updateMetrics();  }### Answer:
@Test public void shouldRecordMessagesConsumed() { int numMessagesConsumed = 500; consumeMessages(numMessagesConsumed, "group1"); Metrics metrics = MetricCollectors.getMetrics(); engineMetrics.updateMetrics(); double value = getMetricValue(metrics, "messages-consumed-per-sec"); assertEquals(numMessagesConsumed / 100, Math.floor(value), 0.01); }
@Test public void shouldRecordMessagesProduced() { int numMessagesProduced = 500; produceMessages(numMessagesProduced); Metrics metrics = MetricCollectors.getMetrics(); engineMetrics.updateMetrics(); double value = getMetricValue(metrics, "messages-produced-per-sec"); assertEquals(numMessagesProduced / 100, Math.floor(value), 0.01); }
@Test public void shouldRecordMessagesConsumedByQuery() { int numMessagesConsumed = 500; consumeMessages(numMessagesConsumed, "group1"); consumeMessages(numMessagesConsumed * 100, "group2"); Metrics metrics = MetricCollectors.getMetrics(); engineMetrics.updateMetrics(); double maxValue = getMetricValue(metrics, "messages-consumed-max"); assertEquals(numMessagesConsumed, Math.floor(maxValue), 5.0); double minValue = getMetricValue(metrics, "messages-consumed-min"); assertEquals(numMessagesConsumed / 100, Math.floor(minValue), 0.01); }