@Test public void shouldReturnEmptyIteratorIfNoData() throws Exception { final WindowStoreIterator<String> iterator = windowStore.fetch("my-key", 0L, 25L); assertEquals(false, iterator.hasNext()); }
public <IteratorType extends KeyValueIterator<?, V>> IteratorType fetch(Fetcher<K, V, IteratorType> fetcher) { final List<ReadOnlyWindowStore<K, V>> stores = provider.stores(storeName, windowStoreType); for (ReadOnlyWindowStore<K, V> windowStore : stores) { try { final IteratorType result = fetcher.fetch(windowStore); if (!result.hasNext()) { result.close(); } else { return result; } } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException( "State store is not available anymore and may have been migrated to another instance; " + "please re-discover its location from the state metadata."); } } return fetcher.empty(); }
CompositeReadOnlyWindowStore implements ReadOnlyWindowStore<K, V> { public <IteratorType extends KeyValueIterator<?, V>> IteratorType fetch(Fetcher<K, V, IteratorType> fetcher) { final List<ReadOnlyWindowStore<K, V>> stores = provider.stores(storeName, windowStoreType); for (ReadOnlyWindowStore<K, V> windowStore : stores) { try { final IteratorType result = fetcher.fetch(windowStore); if (!result.hasNext()) { result.close(); } else { return result; } } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException( "State store is not available anymore and may have been migrated to another instance; " + "please re-discover its location from the state metadata."); } } return fetcher.empty(); } }
CompositeReadOnlyWindowStore implements ReadOnlyWindowStore<K, V> { public <IteratorType extends KeyValueIterator<?, V>> IteratorType fetch(Fetcher<K, V, IteratorType> fetcher) { final List<ReadOnlyWindowStore<K, V>> stores = provider.stores(storeName, windowStoreType); for (ReadOnlyWindowStore<K, V> windowStore : stores) { try { final IteratorType result = fetcher.fetch(windowStore); if (!result.hasNext()) { result.close(); } else { return result; } } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException( "State store is not available anymore and may have been migrated to another instance; " + "please re-discover its location from the state metadata."); } } return fetcher.empty(); } CompositeReadOnlyWindowStore(final StateStoreProvider provider,
                                        final QueryableStoreType<ReadOnlyWindowStore<K, V>> windowStoreType,
                                        final String storeName); }
CompositeReadOnlyWindowStore implements ReadOnlyWindowStore<K, V> { public <IteratorType extends KeyValueIterator<?, V>> IteratorType fetch(Fetcher<K, V, IteratorType> fetcher) { final List<ReadOnlyWindowStore<K, V>> stores = provider.stores(storeName, windowStoreType); for (ReadOnlyWindowStore<K, V> windowStore : stores) { try { final IteratorType result = fetcher.fetch(windowStore); if (!result.hasNext()) { result.close(); } else { return result; } } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException( "State store is not available anymore and may have been migrated to another instance; " + "please re-discover its location from the state metadata."); } } return fetcher.empty(); } CompositeReadOnlyWindowStore(final StateStoreProvider provider,
                                        final QueryableStoreType<ReadOnlyWindowStore<K, V>> windowStoreType,
                                        final String storeName); IteratorType fetch(Fetcher<K, V, IteratorType> fetcher); @Override WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo); @Override KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to, final long timeFrom, final long timeTo); }
CompositeReadOnlyWindowStore implements ReadOnlyWindowStore<K, V> { public <IteratorType extends KeyValueIterator<?, V>> IteratorType fetch(Fetcher<K, V, IteratorType> fetcher) { final List<ReadOnlyWindowStore<K, V>> stores = provider.stores(storeName, windowStoreType); for (ReadOnlyWindowStore<K, V> windowStore : stores) { try { final IteratorType result = fetcher.fetch(windowStore); if (!result.hasNext()) { result.close(); } else { return result; } } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException( "State store is not available anymore and may have been migrated to another instance; " + "please re-discover its location from the state metadata."); } } return fetcher.empty(); } CompositeReadOnlyWindowStore(final StateStoreProvider provider,
                                        final QueryableStoreType<ReadOnlyWindowStore<K, V>> windowStoreType,
                                        final String storeName); IteratorType fetch(Fetcher<K, V, IteratorType> fetcher); @Override WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo); @Override KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to, final long timeFrom, final long timeTo);  }