@Override
    public KeyValueIterator<Windowed<K>, V> all() {
        final NextIteratorFunction<Windowed<K>, V, ReadOnlyWindowStore<K, V>> nextIteratorFunction =
            ReadOnlyWindowStore::all;
        return new DelegatingPeekingKeyValueIterator<>(
            storeName,
            new CompositeKeyValueIterator<>(
                provider.stores(storeName, windowStoreType).iterator(),
                nextIteratorFunction));
    }
@Test
    public void shouldGetAllAcrossStores() {
        final ReadOnlyWindowStoreStub<String, String> secondUnderlying = new
            ReadOnlyWindowStoreStub<>(WINDOW_SIZE);
        stubProviderTwo.addStore(storeName, secondUnderlying);
        underlyingWindowStore.put("a", "a", 0L);
        secondUnderlying.put("b", "b", 10L);
        final List<KeyValue<Windowed<String>, String>> results = StreamsTestUtils.toList(windowStore.all());
        assertThat(results, equalTo(Arrays.asList(
            KeyValue.pair(new Windowed<>("a", new TimeWindow(0, WINDOW_SIZE)), "a"),
            KeyValue.pair(new Windowed<>("b", new TimeWindow(10, 10 + WINDOW_SIZE)), "b"))));
    }