@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 shouldThrowNullPointerExceptionIfFetchingNullKey() {
        assertThrows(NullPointerException.class, () -> sessionStore.fetch(null));
    }