@Test public void testCloseIsIdempotent() throws Exception { streams.close(); final int closeCount = MockMetricsReporter.CLOSE_COUNT.get(); streams.close(); Assert.assertEquals("subsequent close() calls should do nothing", closeCount, MockMetricsReporter.CLOSE_COUNT.get()); }
public void close() { close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS); }
KafkaStreams { public void close() { close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS); } }
KafkaStreams { public void close() { close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS); } 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 close() { close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS); } 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 close() { close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS); } 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);  }