@Test public void shouldFlushEvictedItemsIntoUnderlyingStore() throws Exception { int added = addItemsToCache(); final KeyValueIterator<Bytes, byte[]> iter = underlying.fetch(Bytes.wrap("0".getBytes()), DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP); final KeyValue<Bytes, byte[]> next = iter.next(); assertEquals(DEFAULT_TIMESTAMP, keySchema.segmentTimestamp(next.key)); assertArrayEquals("0".getBytes(), next.value); assertFalse(iter.hasNext()); assertEquals(added - 1, cache.size()); }
@Override public synchronized WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) { validateStoreOpen(); final Bytes keyBytes = Bytes.wrap(serdes.rawKey(key)); final WindowStoreIterator<byte[]> underlyingIterator = underlying.fetch(keyBytes, timeFrom, timeTo); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(keyBytes, timeFrom)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(keyBytes, timeTo)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(keyBytes, keyBytes, timeFrom, timeTo); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator( cacheIterator, hasNextCondition, cacheFunction ); return new MergedSortedCacheWindowStoreIterator<>(filteredCacheIterator, underlyingIterator, new StateSerdes<>(serdes.topic(), Serdes.Long(), serdes.valueSerde())); }
CachingWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V>, CachedStateStore<Windowed<K>, V> { @Override public synchronized WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) { validateStoreOpen(); final Bytes keyBytes = Bytes.wrap(serdes.rawKey(key)); final WindowStoreIterator<byte[]> underlyingIterator = underlying.fetch(keyBytes, timeFrom, timeTo); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(keyBytes, timeFrom)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(keyBytes, timeTo)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(keyBytes, keyBytes, timeFrom, timeTo); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator( cacheIterator, hasNextCondition, cacheFunction ); return new MergedSortedCacheWindowStoreIterator<>(filteredCacheIterator, underlyingIterator, new StateSerdes<>(serdes.topic(), Serdes.Long(), serdes.valueSerde())); } }
CachingWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V>, CachedStateStore<Windowed<K>, V> { @Override public synchronized WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) { validateStoreOpen(); final Bytes keyBytes = Bytes.wrap(serdes.rawKey(key)); final WindowStoreIterator<byte[]> underlyingIterator = underlying.fetch(keyBytes, timeFrom, timeTo); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(keyBytes, timeFrom)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(keyBytes, timeTo)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(keyBytes, keyBytes, timeFrom, timeTo); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator( cacheIterator, hasNextCondition, cacheFunction ); return new MergedSortedCacheWindowStoreIterator<>(filteredCacheIterator, underlyingIterator, new StateSerdes<>(serdes.topic(), Serdes.Long(), serdes.valueSerde())); } CachingWindowStore(final WindowStore<Bytes, byte[]> underlying,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde,
                       final long windowSize,
                       final long segmentInterval); }
CachingWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V>, CachedStateStore<Windowed<K>, V> { @Override public synchronized WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) { validateStoreOpen(); final Bytes keyBytes = Bytes.wrap(serdes.rawKey(key)); final WindowStoreIterator<byte[]> underlyingIterator = underlying.fetch(keyBytes, timeFrom, timeTo); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(keyBytes, timeFrom)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(keyBytes, timeTo)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(keyBytes, keyBytes, timeFrom, timeTo); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator( cacheIterator, hasNextCondition, cacheFunction ); return new MergedSortedCacheWindowStoreIterator<>(filteredCacheIterator, underlyingIterator, new StateSerdes<>(serdes.topic(), Serdes.Long(), serdes.valueSerde())); } CachingWindowStore(final WindowStore<Bytes, byte[]> underlying,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde,
                       final long windowSize,
                       final long segmentInterval); @SuppressWarnings("unchecked") @Override void init(final ProcessorContext context, final StateStore root); void setFlushListener(CacheFlushListener<Windowed<K>, V> flushListener); @Override synchronized void flush(); @Override void close(); @Override synchronized void put(final K key, final V value); @Override synchronized void put(final K key, final V value, final long timestamp); @Override synchronized 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); }
CachingWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V>, CachedStateStore<Windowed<K>, V> { @Override public synchronized WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) { validateStoreOpen(); final Bytes keyBytes = Bytes.wrap(serdes.rawKey(key)); final WindowStoreIterator<byte[]> underlyingIterator = underlying.fetch(keyBytes, timeFrom, timeTo); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(keyBytes, timeFrom)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(keyBytes, timeTo)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(keyBytes, keyBytes, timeFrom, timeTo); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator( cacheIterator, hasNextCondition, cacheFunction ); return new MergedSortedCacheWindowStoreIterator<>(filteredCacheIterator, underlyingIterator, new StateSerdes<>(serdes.topic(), Serdes.Long(), serdes.valueSerde())); } CachingWindowStore(final WindowStore<Bytes, byte[]> underlying,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde,
                       final long windowSize,
                       final long segmentInterval); @SuppressWarnings("unchecked") @Override void init(final ProcessorContext context, final StateStore root); void setFlushListener(CacheFlushListener<Windowed<K>, V> flushListener); @Override synchronized void flush(); @Override void close(); @Override synchronized void put(final K key, final V value); @Override synchronized void put(final K key, final V value, final long timestamp); @Override synchronized 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);  }