@Test
public void testCloseIsIdempotent() throws Exception {

    final Properties props = new Properties();
    props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "testCloseIsIdempotent");
    props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
    props.setProperty(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
    final KStreamBuilder builder = new KStreamBuilder();
    final KafkaStreams streams = new KafkaStreams(builder, props);
    streams.close();
    final int closeCount = MockMetricsReporter.CLOSE_COUNT.get();
    streams.close();
    Assert.assertEquals("subsequent close() calls should do nothing",
        closeCount, MockMetricsReporter.CLOSE_COUNT.get());
}