KafkaStreams { public synchronized void close() { log.debug("Stopping Kafka Stream process"); if (this.state == RUNNING) { for (final StreamThread thread : this.threads) thread.close(); for (final StreamThread thread : this.threads) { try { thread.join(); } catch (final InterruptedException ex) { Thread.interrupted(); } } } if (this.state != STOPPED) { this.metrics.close(); this.state = STOPPED; log.info("Stopped Kafka Stream process"); } } KafkaStreams(final TopologyBuilder builder, final Properties props);  KafkaStreams(final TopologyBuilder builder, final StreamsConfig config);  KafkaStreams(final TopologyBuilder builder, final StreamsConfig config, final KafkaClientSupplier clientSupplier); synchronized void start(); synchronized void close(); void cleanUp(); void setUncaughtExceptionHandler(final Thread.UncaughtExceptionHandler eh);  }
@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()); }