@Test public void testInitialization() throws Exception { KafkaStore<String, String> kafkaStore = StoreUtils.createAndInitKafkaStoreInstance(bootstrapServers); kafkaStore.close(); }
@Override public void close() { try { if (kafkaTopicReader != null) { kafkaTopicReader.shutdown(); log.debug("Kafka store reader thread shut down"); } if (producer != null) { producer.close(); log.debug("Kafka store producer shut down"); } localStore.close(); if (storeUpdateHandler != null) { storeUpdateHandler.close(); } log.debug("Kafka store shut down complete"); } catch (Exception e) { throw new RuntimeException(e); } }
KafkaStore implements Store<K, V> { @Override public void close() { try { if (kafkaTopicReader != null) { kafkaTopicReader.shutdown(); log.debug("Kafka store reader thread shut down"); } if (producer != null) { producer.close(); log.debug("Kafka store producer shut down"); } localStore.close(); if (storeUpdateHandler != null) { storeUpdateHandler.close(); } log.debug("Kafka store shut down complete"); } catch (Exception e) { throw new RuntimeException(e); } } }
KafkaStore implements Store<K, V> { @Override public void close() { try { if (kafkaTopicReader != null) { kafkaTopicReader.shutdown(); log.debug("Kafka store reader thread shut down"); } if (producer != null) { producer.close(); log.debug("Kafka store producer shut down"); } localStore.close(); if (storeUpdateHandler != null) { storeUpdateHandler.close(); } log.debug("Kafka store shut down complete"); } catch (Exception e) { throw new RuntimeException(e); } } KafkaStore(SchemaRegistryConfig config,
                    StoreUpdateHandler<K, V> storeUpdateHandler,
                    Serializer<K, V> serializer,
                    Store<K, V> localStore,
                    K noopKey); }
KafkaStore implements Store<K, V> { @Override public void close() { try { if (kafkaTopicReader != null) { kafkaTopicReader.shutdown(); log.debug("Kafka store reader thread shut down"); } if (producer != null) { producer.close(); log.debug("Kafka store producer shut down"); } localStore.close(); if (storeUpdateHandler != null) { storeUpdateHandler.close(); } log.debug("Kafka store shut down complete"); } catch (Exception e) { throw new RuntimeException(e); } } KafkaStore(SchemaRegistryConfig config,
                    StoreUpdateHandler<K, V> storeUpdateHandler,
                    Serializer<K, V> serializer,
                    Store<K, V> localStore,
                    K noopKey); @Override void init(); static void addSchemaRegistryConfigsToClientProperties(SchemaRegistryConfig config,
                                                                Properties props); void waitUntilKafkaReaderReachesLastOffset(int timeoutMs); void waitUntilKafkaReaderReachesLastOffset(String subject, int timeoutMs); void markLastWrittenOffsetInvalid(); @Override V get(K key); @Override V put(K key, V value); @Override CloseableIterator<V> getAll(K key1, K key2); @Override void putAll(Map<K, V> entries); @Override V delete(K key); @Override CloseableIterator<K> getAllKeys(); @Override void flush(); @Override void close(); void waitForInit(); boolean initialized(); long lastOffset(String subject); void setLastOffset(String subject, long lastOffset); Lock leaderLock(); Lock lockFor(String subject); }
KafkaStore implements Store<K, V> { @Override public void close() { try { if (kafkaTopicReader != null) { kafkaTopicReader.shutdown(); log.debug("Kafka store reader thread shut down"); } if (producer != null) { producer.close(); log.debug("Kafka store producer shut down"); } localStore.close(); if (storeUpdateHandler != null) { storeUpdateHandler.close(); } log.debug("Kafka store shut down complete"); } catch (Exception e) { throw new RuntimeException(e); } } KafkaStore(SchemaRegistryConfig config,
                    StoreUpdateHandler<K, V> storeUpdateHandler,
                    Serializer<K, V> serializer,
                    Store<K, V> localStore,
                    K noopKey); @Override void init(); static void addSchemaRegistryConfigsToClientProperties(SchemaRegistryConfig config,
                                                                Properties props); void waitUntilKafkaReaderReachesLastOffset(int timeoutMs); void waitUntilKafkaReaderReachesLastOffset(String subject, int timeoutMs); void markLastWrittenOffsetInvalid(); @Override V get(K key); @Override V put(K key, V value); @Override CloseableIterator<V> getAll(K key1, K key2); @Override void putAll(Map<K, V> entries); @Override V delete(K key); @Override CloseableIterator<K> getAllKeys(); @Override void flush(); @Override void close(); void waitForInit(); boolean initialized(); long lastOffset(String subject); void setLastOffset(String subject, long lastOffset); Lock leaderLock(); Lock lockFor(String subject);  }