@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(); }
public void cleanUp() { if (state.isRunning()) { throw new IllegalStateException("Cannot clean up while running."); } final String appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final String stateDir = config.getString(StreamsConfig.STATE_DIR_CONFIG); final String localApplicationDir = stateDir + File.separator + appId; log.debug("{} Removing local Kafka Streams application data in {} for application {}.", logPrefix, localApplicationDir, appId); final StateDirectory stateDirectory = new StateDirectory(appId, "cleanup", stateDir, Time.SYSTEM); stateDirectory.cleanRemovedTasks(0); }
KafkaStreams { public void cleanUp() { if (state.isRunning()) { throw new IllegalStateException("Cannot clean up while running."); } final String appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final String stateDir = config.getString(StreamsConfig.STATE_DIR_CONFIG); final String localApplicationDir = stateDir + File.separator + appId; log.debug("{} Removing local Kafka Streams application data in {} for application {}.", logPrefix, localApplicationDir, appId); final StateDirectory stateDirectory = new StateDirectory(appId, "cleanup", stateDir, Time.SYSTEM); stateDirectory.cleanRemovedTasks(0); } }
KafkaStreams { public void cleanUp() { if (state.isRunning()) { throw new IllegalStateException("Cannot clean up while running."); } final String appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final String stateDir = config.getString(StreamsConfig.STATE_DIR_CONFIG); final String localApplicationDir = stateDir + File.separator + appId; log.debug("{} Removing local Kafka Streams application data in {} for application {}.", logPrefix, localApplicationDir, appId); final StateDirectory stateDirectory = new StateDirectory(appId, "cleanup", stateDir, Time.SYSTEM); stateDirectory.cleanRemovedTasks(0); } KafkaStreams(final TopologyBuilder builder, final Properties props);  KafkaStreams(final TopologyBuilder builder, final StreamsConfig config);  KafkaStreams(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier); }
KafkaStreams { public void cleanUp() { if (state.isRunning()) { throw new IllegalStateException("Cannot clean up while running."); } final String appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final String stateDir = config.getString(StreamsConfig.STATE_DIR_CONFIG); final String localApplicationDir = stateDir + File.separator + appId; log.debug("{} Removing local Kafka Streams application data in {} for application {}.", logPrefix, localApplicationDir, appId); final StateDirectory stateDirectory = new StateDirectory(appId, "cleanup", stateDir, Time.SYSTEM); stateDirectory.cleanRemovedTasks(0); } KafkaStreams(final TopologyBuilder builder, final Properties props);  KafkaStreams(final TopologyBuilder builder, final StreamsConfig config);  KafkaStreams(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier); void setStateListener(final KafkaStreams.StateListener listener); synchronized State state(); Map<MetricName, ? extends Metric> metrics(); synchronized void start(); void close(); synchronized boolean close(final long timeout, final TimeUnit timeUnit); @Override String toString(); String toString(final String indent); void cleanUp(); void setUncaughtExceptionHandler(final Thread.UncaughtExceptionHandler eh); Collection<StreamsMetadata> allMetadata(); Collection<StreamsMetadata> allMetadataForStore(final String storeName); StreamsMetadata metadataForKey(final String storeName,
                                              final K key,
                                              final Serializer<K> keySerializer); StreamsMetadata metadataForKey(final String storeName,
                                              final K key,
                                              final StreamPartitioner<? super K, ?> partitioner); T store(final String storeName, final QueryableStoreType<T> queryableStoreType); }
KafkaStreams { public void cleanUp() { if (state.isRunning()) { throw new IllegalStateException("Cannot clean up while running."); } final String appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final String stateDir = config.getString(StreamsConfig.STATE_DIR_CONFIG); final String localApplicationDir = stateDir + File.separator + appId; log.debug("{} Removing local Kafka Streams application data in {} for application {}.", logPrefix, localApplicationDir, appId); final StateDirectory stateDirectory = new StateDirectory(appId, "cleanup", stateDir, Time.SYSTEM); stateDirectory.cleanRemovedTasks(0); } KafkaStreams(final TopologyBuilder builder, final Properties props);  KafkaStreams(final TopologyBuilder builder, final StreamsConfig config);  KafkaStreams(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier); void setStateListener(final KafkaStreams.StateListener listener); synchronized State state(); Map<MetricName, ? extends Metric> metrics(); synchronized void start(); void close(); synchronized boolean close(final long timeout, final TimeUnit timeUnit); @Override String toString(); String toString(final String indent); void cleanUp(); void setUncaughtExceptionHandler(final Thread.UncaughtExceptionHandler eh); Collection<StreamsMetadata> allMetadata(); Collection<StreamsMetadata> allMetadataForStore(final String storeName); StreamsMetadata metadataForKey(final String storeName,
                                              final K key,
                                              final Serializer<K> keySerializer); StreamsMetadata metadataForKey(final String storeName,
                                              final K key,
                                              final StreamPartitioner<? super K, ?> partitioner); T store(final String storeName, final QueryableStoreType<T> queryableStoreType);  }