@Test public void shouldNotGetValueFromOtherStores() throws Exception { 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); final KeyValueIterator<Windowed<String>, Long> result = sessionStore.fetch("foo"); assertEquals(KeyValue.pair(expectedKey, 1L), result.next()); assertFalse(result.hasNext()); }
private KeyValueIterator<Windowed<K>, V> fetch(Fetcher<K, V> fetcher) { final List<ReadOnlySessionStore<K, V>> stores = storeProvider.stores(storeName, queryableStoreType); for (final ReadOnlySessionStore<K, V> store : stores) { try { final KeyValueIterator<Windowed<K>, V> result = fetcher.fetch(store); 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."); } } return KeyValueIterators.emptyIterator(); }
CompositeReadOnlySessionStore implements ReadOnlySessionStore<K, V> { private KeyValueIterator<Windowed<K>, V> fetch(Fetcher<K, V> fetcher) { final List<ReadOnlySessionStore<K, V>> stores = storeProvider.stores(storeName, queryableStoreType); for (final ReadOnlySessionStore<K, V> store : stores) { try { final KeyValueIterator<Windowed<K>, V> result = fetcher.fetch(store); 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."); } } return KeyValueIterators.emptyIterator(); } }
CompositeReadOnlySessionStore implements ReadOnlySessionStore<K, V> { private KeyValueIterator<Windowed<K>, V> fetch(Fetcher<K, V> fetcher) { final List<ReadOnlySessionStore<K, V>> stores = storeProvider.stores(storeName, queryableStoreType); for (final ReadOnlySessionStore<K, V> store : stores) { try { final KeyValueIterator<Windowed<K>, V> result = fetcher.fetch(store); 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."); } } return KeyValueIterators.emptyIterator(); } CompositeReadOnlySessionStore(final StateStoreProvider storeProvider,
                                         final QueryableStoreType<ReadOnlySessionStore<K, V>> queryableStoreType,
                                         final String storeName); }
CompositeReadOnlySessionStore implements ReadOnlySessionStore<K, V> { private KeyValueIterator<Windowed<K>, V> fetch(Fetcher<K, V> fetcher) { final List<ReadOnlySessionStore<K, V>> stores = storeProvider.stores(storeName, queryableStoreType); for (final ReadOnlySessionStore<K, V> store : stores) { try { final KeyValueIterator<Windowed<K>, V> result = fetcher.fetch(store); 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."); } } return KeyValueIterators.emptyIterator(); } CompositeReadOnlySessionStore(final StateStoreProvider storeProvider,
                                         final QueryableStoreType<ReadOnlySessionStore<K, V>> queryableStoreType,
                                         final String storeName); @Override KeyValueIterator<Windowed<K>, V> fetch(final K key); @Override KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to); }
CompositeReadOnlySessionStore implements ReadOnlySessionStore<K, V> { private KeyValueIterator<Windowed<K>, V> fetch(Fetcher<K, V> fetcher) { final List<ReadOnlySessionStore<K, V>> stores = storeProvider.stores(storeName, queryableStoreType); for (final ReadOnlySessionStore<K, V> store : stores) { try { final KeyValueIterator<Windowed<K>, V> result = fetcher.fetch(store); 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."); } } return KeyValueIterators.emptyIterator(); } CompositeReadOnlySessionStore(final StateStoreProvider storeProvider,
                                         final QueryableStoreType<ReadOnlySessionStore<K, V>> queryableStoreType,
                                         final String storeName); @Override KeyValueIterator<Windowed<K>, V> fetch(final K key); @Override KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to);  }