@Test
public void testRun() throws Exception {

    val outlierMmdJson = objectMapper.writeValueAsString(outlierMMD);
    val breakoutMmdJson = objectMapper.writeValueAsString(breakoutMMD);
    for (int i = 0; i < NUM_OUTLIER_METRICS; i++) {
        log.info("Writing mapped metric data: {}", outlierMmdJson);
        kafka.helper().produceStrings(METRIC_TOPIC, outlierMmdJson);
    }
    for (int i = 0; i < NUM_BREAKOUT_METRICS; i++) {
        log.info("Writing mapped metric data: {}", breakoutMmdJson);
        kafka.helper().produceStrings(METRIC_TOPIC, breakoutMmdJson);
    }
    val managerThread = new Thread(managerUnderTest);
    managerThread.start();
    managerThread.join(THREAD_JOIN_MILLIS);
    managerUnderTest.getMetricConsumer().wakeup();
    // FIXME This hangs if there aren't enough messages to consume. Add a timeout?
    val outliers = kafka.helper().consumeStrings(OUTLIER_TOPIC, NUM_OUTLIER_METRICS).get();
    assertEquals(NUM_OUTLIER_METRICS, outliers.size());
    val breakouts = kafka.helper().consumeStrings(BREAKOUT_TOPIC, NUM_BREAKOUT_METRICS).get();
    assertEquals(NUM_BREAKOUT_METRICS, breakouts.size());
    for (val outlier : outliers) {
        log.info("outlierDetectorResult={}", outlier);
    }
}