### Question:
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);  }### Answer:
@Test public void shouldFetchResulstFromUnderlyingSessionStore() throws Exception { underlyingSessionStore.put(new Windowed<>("a", new SessionWindow(0, 0)), 1L); underlyingSessionStore.put(new Windowed<>("a", new SessionWindow(10, 10)), 2L); final List<KeyValue<Windowed<String>, Long>> results = toList(sessionStore.fetch("a")); assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), 1L), KeyValue.pair(new Windowed<>("a", new SessionWindow(10, 10)), 2L)), results); }
@Test public void shouldReturnEmptyIteratorIfNoData() throws Exception { final KeyValueIterator<Windowed<String>, Long> result = sessionStore.fetch("b"); assertFalse(result.hasNext()); }
@Test public void shouldFindValueForKeyWhenMultiStores() throws Exception { final ReadOnlySessionStoreStub<String, Long> secondUnderlying = new ReadOnlySessionStoreStub<>(); stubProviderTwo.addStore(storeName, secondUnderlying); final Windowed<String> keyOne = new Windowed<>("key-one", new SessionWindow(0, 0)); final Windowed<String> keyTwo = new Windowed<>("key-two", new SessionWindow(0, 0)); underlyingSessionStore.put(keyOne, 0L); secondUnderlying.put(keyTwo, 10L); final List<KeyValue<Windowed<String>, Long>> keyOneResults = toList(sessionStore.fetch("key-one")); final List<KeyValue<Windowed<String>, Long>> keyTwoResults = toList(sessionStore.fetch("key-two")); assertEquals(Collections.singletonList(KeyValue.pair(keyOne, 0L)), keyOneResults); assertEquals(Collections.singletonList(KeyValue.pair(keyTwo, 10L)), keyTwoResults); }
@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()); }
@Test(expected = InvalidStateStoreException.class) public void shouldThrowInvalidStateStoreExceptionOnRebalance() throws Exception { final CompositeReadOnlySessionStore<String, String> store = new CompositeReadOnlySessionStore<>(new StateStoreProviderStub(true), QueryableStoreTypes.<String, String>sessionStore(), "whateva"); store.fetch("a"); }
@Test(expected = InvalidStateStoreException.class) public void shouldThrowInvalidStateStoreExceptionIfFetchThrows() throws Exception { underlyingSessionStore.setOpen(false); underlyingSessionStore.fetch("key"); }