@Override
    public KeyValueIterator<Windowed<K>, V> fetch(final K key) {
        Objects.requireNonNull(key, "key can't be null");
        final List<ReadOnlySessionStore<K, V>> stores = storeProvider.stores(storeName, queryableStoreType);
        for (final ReadOnlySessionStore<K, V> store : stores) {
            try {
                final KeyValueIterator<Windowed<K>, V> result = store.fetch(key);
                if (!result.hasNext()) {
                    result.close();
                } else {
                    return result;
                }
            } catch (final InvalidStateStoreException ise) {
                throw new InvalidStateStoreException("State store  [" + storeName + "] is not available anymore" +
                                                             " and may have been migrated to another instance; " +
                                                             "please re-discover its location from the state metadata. " +
                                                             "Original error message: " + ise);
            }
        }
        return KeyValueIterators.emptyIterator();
    }
@Test
    public void shouldNotGetValueFromOtherStores() {
        final Windowed<String> expectedKey = new Windowed<>("foo", new SessionWindow(0, 0));
        otherUnderlyingStore.put(new Windowed<>("foo", new SessionWindow(10, 10)), 10L);
        underlyingSessionStore.put(expectedKey, 1L);

        try (final KeyValueIterator<Windowed<String>, Long> result = sessionStore.fetch("foo")) {
            assertEquals(KeyValue.pair(expectedKey, 1L), result.next());
            assertFalse(result.hasNext());
        }
    }