KafkaStore extends ForwardingMap<K, V> implements Map<K, V>, AutoCloseable { @Override public V put(K key, V value) { try { putInternal(key, value); sync(); return value; } catch (InterruptedException e) { throw new RuntimeException(e); } } KafkaStore(String bootstrapServers, Serializer<K, V> serializer, String topic);  KafkaStore(
      String bootstrapServers,
      Serializer<K, V> serializer,
      String topic,
      Collection<StoreUpdateObserver<K, V>> observers);  KafkaStore(
      String bootstrapServers,
      Serializer<K, V> serializer,
      String topic,
      Time time,
      Map<String, Object> additionalProducerProps,
      Map<String, Object> additionalConsumerProps);  KafkaStore(
      String bootstrapServers,
      Serializer<K, V> serializer,
      String topic,
      Map<String, Object> additionalProducerProps,
      Map<String, Object> additionalConsumerProps);  KafkaStore(
      String bootstrapServers,
      Serializer<K, V> serializer,
      String topic,
      Collection<StoreUpdateObserver<K, V>> observers,
      Time time,
      Map<String, Object> additionalProducerProps,
      Map<String, Object> additionalConsumerProps); @Override V put(K key, V value); @Override void putAll(Map<? extends K, ? extends V> m); @SuppressWarnings("unchecked") @Override V remove(Object key); @Override void clear(); @Override void close();  }
@Test public void setAndThenGetValue() throws Exception { String key = UUID.randomUUID().toString(); try (KafkaStore<String, String> store = new KafkaStore<>(CLUSTER.bootstrapServers(), SERIALIZER, TOPIC)) { store.put(key, "test-value"); String value = store.get(key); assertThat(value, is("test-value")); } }
@Test public void valueStillAvailableAfterOriginalStoreCloses() throws Exception { String key = UUID.randomUUID().toString(); try (KafkaStore<String, String> store = new KafkaStore<>(CLUSTER.bootstrapServers(), SERIALIZER, TOPIC)) { store.put(key, "test-value"); String value = store.get(key); assertThat(value, is("test-value")); } try (KafkaStore<String, String> store = new KafkaStore<>(CLUSTER.bootstrapServers(), SERIALIZER, TOPIC)) { String value = store.get(key); assertThat(value, is("test-value")); } }
@Test public void nullKeyIsStored() throws Exception { try (KafkaStore<String, String> store = new KafkaStore<>(CLUSTER.bootstrapServers(), SERIALIZER, TOPIC)) { store.put(null, "value"); assertThat(store.get(null), is("value")); assertThat(store.containsKey(null), is(true)); } }
@Test public void nullValueIsAccepted() throws Exception { try (KafkaStore<String, String> store = new KafkaStore<>(CLUSTER.bootstrapServers(), SERIALIZER, TOPIC)) { store.put("key", null); assertThat(store.get("key"), is(nullValue())); assertThat(store.containsKey("key"), is(true)); } }