KafkaStreams { public void cleanUp() { if (this.state == RUNNING) { throw new IllegalStateException("Cannot clean up while running."); } final String localApplicationDir = this.config.getString(StreamsConfig.STATE_DIR_CONFIG) + File.separator + this.config.getString(StreamsConfig.APPLICATION_ID_CONFIG); log.debug("Clean up local Kafka Streams data in {}", localApplicationDir); log.debug("Removing local Kafka Streams application data in {} for application {}", localApplicationDir, this.config.getString(StreamsConfig.APPLICATION_ID_CONFIG)); Utils.delete(new File(localApplicationDir)); } 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 testCleanup() throws Exception { final Properties props = new Properties(); props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "testLocalCleanup"); props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); final KStreamBuilder builder = new KStreamBuilder(); final KafkaStreams streams = new KafkaStreams(builder, props); streams.cleanUp(); streams.start(); streams.close(); streams.cleanUp(); }
@Test public void testCleanupIsolation() throws Exception { final KStreamBuilder builder = new KStreamBuilder(); final String appId1 = "testIsolation-1"; final String appId2 = "testIsolation-2"; final String stateDir = TestUtils.tempDirectory("kafka-test").getPath(); final File stateDirApp1 = new File(stateDir + File.separator + appId1); final File stateDirApp2 = new File(stateDir + File.separator + appId2); final Properties props = new Properties(); props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); props.put(StreamsConfig.STATE_DIR_CONFIG, stateDir); assertFalse(stateDirApp1.exists()); assertFalse(stateDirApp2.exists()); props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId1); final KafkaStreams streams1 = new KafkaStreams(builder, props); props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId2); final KafkaStreams streams2 = new KafkaStreams(builder, props); assertTrue(stateDirApp1.exists()); assertTrue(stateDirApp2.exists()); streams1.cleanUp(); assertFalse(stateDirApp1.exists()); assertTrue(stateDirApp2.exists()); streams2.cleanUp(); assertFalse(stateDirApp1.exists()); assertFalse(stateDirApp2.exists()); }