@Test(expected = InvalidStateStoreException.class) public void shouldThrowInvalidStoreExceptionOnAllDuringRebalance() throws Exception { rebalancing().all(); }
@Override public KeyValueIterator<K, V> all() { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.all(); } 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."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> all() { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.all(); } 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."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> all() { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.all(); } 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."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } CompositeReadOnlyKeyValueStore(final StateStoreProvider storeProvider,
                                          final QueryableStoreType<ReadOnlyKeyValueStore<K, V>> storeType,
                                          final String storeName); }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> all() { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.all(); } 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."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } CompositeReadOnlyKeyValueStore(final StateStoreProvider storeProvider,
                                          final QueryableStoreType<ReadOnlyKeyValueStore<K, V>> storeType,
                                          final String storeName); @Override V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override long approximateNumEntries(); }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> all() { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.all(); } 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."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } CompositeReadOnlyKeyValueStore(final StateStoreProvider storeProvider,
                                          final QueryableStoreType<ReadOnlyKeyValueStore<K, V>> storeType,
                                          final String storeName); @Override V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override long approximateNumEntries();  }