### Question:
CachingKeyValueStore extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V>, CachedStateStore<K, V> { @Override public KeyValueIterator<K, V> all() { validateStoreOpen(); final KeyValueIterator<Bytes, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>(this.name(), underlying.all()); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.all(cacheName); return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes); } CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying,
                         final Serde<K> keySerde,
                         final Serde<V> valueSerde); @SuppressWarnings("unchecked") @Override void init(final ProcessorContext context, final StateStore root); void setFlushListener(final CacheFlushListener<K, V> flushListener); @Override synchronized void flush(); @Override void close(); @Override boolean persistent(); @Override boolean isOpen(); @Override synchronized V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override synchronized long approximateNumEntries(); @Override synchronized void put(final K key, final V value); @Override synchronized V putIfAbsent(final K key, final V value); @Override synchronized void putAll(final List<KeyValue<K, V>> entries); @Override synchronized V delete(final K key); @Override StateStore inner();  }### Answer:
@Test public void shouldIterateAllStoredItems() throws Exception { int items = addItemsToCache(); final KeyValueIterator<String, String> all = store.all(); final List<String> results = new ArrayList<>(); while (all.hasNext()) { results.add(all.next().key); } assertEquals(items, results.size()); }