diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java index 07337806922cb..aa7b2cc17baa8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java @@ -33,20 +33,18 @@ public class InMemoryKeyValueStore implements KeyValueStore { private final String name; - private final ConcurrentNavigableMap map; + private final ConcurrentNavigableMap map = new ConcurrentSkipListMap<>(); private volatile boolean open = false; private static final Logger LOG = LoggerFactory.getLogger(InMemoryKeyValueStore.class); public InMemoryKeyValueStore(final String name) { this.name = name; - - this.map = new ConcurrentSkipListMap<>(); } @Override public String name() { - return this.name; + return name; } @Override @@ -65,7 +63,7 @@ public void init(final ProcessorContext context, }); } - this.open = true; + open = true; } @Override @@ -75,20 +73,20 @@ public boolean persistent() { @Override public boolean isOpen() { - return this.open; + return open; } @Override public byte[] get(final Bytes key) { - return this.map.get(key); + return map.get(key); } @Override public void put(final Bytes key, final byte[] value) { if (value == null) { - this.map.remove(key); + map.remove(key); } else { - this.map.put(key, value); + map.put(key, value); } } @@ -110,7 +108,7 @@ public void putAll(final List> entries) { @Override public byte[] delete(final Bytes key) { - return this.map.remove(key); + return map.remove(key); } @Override @@ -125,19 +123,19 @@ public KeyValueIterator range(final Bytes from, final Bytes to) { return new DelegatingPeekingKeyValueIterator<>( name, - new InMemoryKeyValueIterator(this.map.subMap(from, true, to, true).entrySet().iterator())); + new InMemoryKeyValueIterator(map.subMap(from, true, to, true).entrySet().iterator())); } @Override public KeyValueIterator all() { return new DelegatingPeekingKeyValueIterator<>( name, - new InMemoryKeyValueIterator(this.map.entrySet().iterator())); + new InMemoryKeyValueIterator(map.entrySet().iterator())); } @Override public long approximateNumEntries() { - return this.map.size(); + return map.size(); } @Override @@ -147,8 +145,8 @@ public void flush() { @Override public void close() { - this.map.clear(); - this.open = false; + map.clear(); + open = false; } private static class InMemoryKeyValueIterator implements KeyValueIterator { diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java index 0cee668e2a8d6..797a5d9a2baec 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java @@ -63,8 +63,8 @@ public class InMemoryWindowStore implements WindowStore { private final long windowSize; private final boolean retainDuplicates; - private final ConcurrentNavigableMap> segmentMap; - private final Set openIterators; + private final ConcurrentNavigableMap> segmentMap = new ConcurrentSkipListMap<>(); + private final Set openIterators = ConcurrentHashMap.newKeySet(); private volatile boolean open = false; @@ -78,18 +78,14 @@ public class InMemoryWindowStore implements WindowStore { this.windowSize = windowSize; this.retainDuplicates = retainDuplicates; this.metricScope = metricScope; - - this.openIterators = ConcurrentHashMap.newKeySet(); - this.segmentMap = new ConcurrentSkipListMap<>(); } @Override public String name() { - return this.name; + return name; } @Override - @SuppressWarnings("unchecked") public void init(final ProcessorContext context, final StateStore root) { this.context = (InternalProcessorContext) context; @@ -113,7 +109,7 @@ public void init(final ProcessorContext context, final StateStore root) { put(Bytes.wrap(extractStoreKeyBytes(key)), value, extractStoreTimestamp(key)); }); } - this.open = true; + open = true; } @Override @@ -125,19 +121,19 @@ public void put(final Bytes key, final byte[] value) { public void put(final Bytes key, final byte[] value, final long windowStartTimestamp) { removeExpiredSegments(); maybeUpdateSeqnumForDups(); - this.observedStreamTime = Math.max(this.observedStreamTime, windowStartTimestamp); + observedStreamTime = Math.max(observedStreamTime, windowStartTimestamp); final Bytes keyBytes = retainDuplicates ? wrapForDups(key, seqnum) : key; - if (windowStartTimestamp <= this.observedStreamTime - this.retentionPeriod) { + if (windowStartTimestamp <= observedStreamTime - retentionPeriod) { expiredRecordSensor.record(); LOG.warn("Skipping record for expired segment."); } else { if (value != null) { - this.segmentMap.computeIfAbsent(windowStartTimestamp, t -> new ConcurrentSkipListMap<>()); - this.segmentMap.get(windowStartTimestamp).put(keyBytes, value); + segmentMap.computeIfAbsent(windowStartTimestamp, t -> new ConcurrentSkipListMap<>()); + segmentMap.get(windowStartTimestamp).put(keyBytes, value); } else { - this.segmentMap.computeIfPresent(windowStartTimestamp, (t, kvMap) -> { + segmentMap.computeIfPresent(windowStartTimestamp, (t, kvMap) -> { kvMap.remove(keyBytes); return kvMap; }); @@ -149,11 +145,11 @@ public void put(final Bytes key, final byte[] value, final long windowStartTimes public byte[] fetch(final Bytes key, final long windowStartTimestamp) { removeExpiredSegments(); - if (windowStartTimestamp <= this.observedStreamTime - this.retentionPeriod) { + if (windowStartTimestamp <= observedStreamTime - retentionPeriod) { return null; } - final ConcurrentNavigableMap kvMap = this.segmentMap.get(windowStartTimestamp); + final ConcurrentNavigableMap kvMap = segmentMap.get(windowStartTimestamp); if (kvMap == null) { return null; } else { @@ -167,14 +163,14 @@ public WindowStoreIterator fetch(final Bytes key, final long timeFrom, f removeExpiredSegments(); // add one b/c records expire exactly retentionPeriod ms after created - final long minTime = Math.max(timeFrom, this.observedStreamTime - this.retentionPeriod + 1); + final long minTime = Math.max(timeFrom, observedStreamTime - retentionPeriod + 1); if (timeTo < minTime) { - return new WrappedInMemoryWindowStoreIterator(); + return WrappedInMemoryWindowStoreIterator.emptyIterator(); } - return new WrappedInMemoryWindowStoreIterator( - key, key, this.segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + return registerNewWindowStoreIterator( + key, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @Deprecated @@ -193,14 +189,14 @@ public KeyValueIterator, byte[]> fetch(final Bytes from, } // add one b/c records expire exactly retentionPeriod ms after created - final long minTime = Math.max(timeFrom, this.observedStreamTime - this.retentionPeriod + 1); + final long minTime = Math.max(timeFrom, observedStreamTime - retentionPeriod + 1); if (timeTo < minTime) { - return new WrappedWindowedKeyValueIterator(); + return KeyValueIterators.emptyIterator(); } - return new WrappedWindowedKeyValueIterator( - from, to, this.segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + return registerNewWindowedKeyValueIterator( + from, to, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @Deprecated @@ -209,24 +205,24 @@ public KeyValueIterator, byte[]> fetchAll(final long timeFrom, f removeExpiredSegments(); // add one b/c records expire exactly retentionPeriod ms after created - final long minTime = Math.max(timeFrom, this.observedStreamTime - this.retentionPeriod + 1); + final long minTime = Math.max(timeFrom, observedStreamTime - retentionPeriod + 1); if (timeTo < minTime) { - return new WrappedWindowedKeyValueIterator(); + return KeyValueIterators.emptyIterator(); } - return new WrappedWindowedKeyValueIterator( - null, null, this.segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + return registerNewWindowedKeyValueIterator( + null, null, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @Override public KeyValueIterator, byte[]> all() { removeExpiredSegments(); - final long minTime = this.observedStreamTime - this.retentionPeriod; + final long minTime = observedStreamTime - retentionPeriod; - return new WrappedWindowedKeyValueIterator( - null, null, this.segmentMap.tailMap(minTime, false).entrySet().iterator()); + return registerNewWindowedKeyValueIterator( + null, null, segmentMap.tailMap(minTime, false).entrySet().iterator()); } @Override @@ -236,7 +232,7 @@ public boolean persistent() { @Override public boolean isOpen() { - return this.open; + return open; } @Override @@ -246,16 +242,16 @@ public void flush() { @Override public void close() { - this.segmentMap.clear(); - this.open = false; + segmentMap.clear(); + open = false; } private void removeExpiredSegments() { - long minLiveTime = Math.max(0L, this.observedStreamTime - this.retentionPeriod + 1); + long minLiveTime = Math.max(0L, observedStreamTime - retentionPeriod + 1); for (final InMemoryWindowStoreIteratorWrapper it : openIterators) { minLiveTime = Math.min(minLiveTime, it.minTime()); } - this.segmentMap.headMap(minLiveTime, false).clear(); + segmentMap.headMap(minLiveTime, false).clear(); } private void maybeUpdateSeqnumForDups() { @@ -279,7 +275,41 @@ private static Bytes getKey(final Bytes keyBytes) { } - private abstract class InMemoryWindowStoreIteratorWrapper implements Comparable { + private WrappedInMemoryWindowStoreIterator registerNewWindowStoreIterator(final Bytes key, + final Iterator>> segmentIterator) { + final Bytes keyFrom = retainDuplicates ? wrapForDups(key, 0) : key; + final Bytes keyTo = retainDuplicates ? wrapForDups(key, Integer.MAX_VALUE) : key; + + final WrappedInMemoryWindowStoreIterator iterator = + new WrappedInMemoryWindowStoreIterator(keyFrom, keyTo, segmentIterator, openIterators::remove); + + openIterators.add(iterator); + return iterator; + } + + private WrappedWindowedKeyValueIterator registerNewWindowedKeyValueIterator(final Bytes keyFrom, + final Bytes keyTo, + final Iterator>> segmentIterator) { + final Bytes from = (retainDuplicates && keyFrom != null) ? wrapForDups(keyFrom, 0) : keyFrom; + final Bytes to = (retainDuplicates && keyTo != null) ? wrapForDups(keyTo, Integer.MAX_VALUE) : keyTo; + + final WrappedWindowedKeyValueIterator iterator = + new WrappedWindowedKeyValueIterator(from, + to, + segmentIterator, + openIterators::remove, + retainDuplicates, + windowSize); + openIterators.add(iterator); + return iterator; + } + + + interface ClosingCallback { + void deregisterIterator(final InMemoryWindowStoreIteratorWrapper iterator); + } + + private static abstract class InMemoryWindowStoreIteratorWrapper { private Iterator>> segmentIterator; private Iterator> recordIterator; @@ -287,31 +317,21 @@ private abstract class InMemoryWindowStoreIteratorWrapper implements Comparable< private long currentTime; private final boolean allKeys; - private Bytes keyFrom; - private Bytes keyTo; - - // Default constructor sets up a dummy iterator when no results are returned (eg entire fetch range is expired) - InMemoryWindowStoreIteratorWrapper() { - this.allKeys = false; - recordIterator = null; - } + private final Bytes keyFrom; + private final Bytes keyTo; + private final ClosingCallback callback; InMemoryWindowStoreIteratorWrapper(final Bytes keyFrom, final Bytes keyTo, - final Iterator>> segmentIterator) { - this.allKeys = (keyFrom == null) && (keyTo == null); - if (retainDuplicates && !allKeys) { - this.keyFrom = wrapForDups(keyFrom, 0); - this.keyTo = wrapForDups(keyTo, Integer.MAX_VALUE); - } else { - this.keyFrom = keyFrom; - this.keyTo = keyTo; - } + final Iterator>> segmentIterator, + final ClosingCallback callback) { + this.keyFrom = keyFrom; + this.keyTo = keyTo; + allKeys = (keyFrom == null) && (keyTo == null); this.segmentIterator = segmentIterator; - this.recordIterator = setRecordIterator(); - - openIterators.add(this); + this.callback = callback; + recordIterator = segmentIterator == null ? null : setRecordIterator(); } public boolean hasNext() { @@ -332,7 +352,7 @@ public void remove() { } public void close() { - openIterators.remove(this); + callback.deregisterIterator(this); } // getNext is only called when either recordIterator or segmentIterator has a next @@ -368,22 +388,15 @@ Iterator> setRecordIterator() { Long minTime() { return currentTime; } - - public int compareTo(final InMemoryWindowStoreIteratorWrapper other) { - return (int) (minTime() - other.minTime()); - } } - private class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIteratorWrapper implements WindowStoreIterator { - - WrappedInMemoryWindowStoreIterator() { - super(); - } + private static class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIteratorWrapper implements WindowStoreIterator { WrappedInMemoryWindowStoreIterator(final Bytes keyFrom, final Bytes keyTo, - final Iterator>> segmentIterator) { - super(keyFrom, keyTo, segmentIterator); + final Iterator>> segmentIterator, + final ClosingCallback callback) { + super(keyFrom, keyTo, segmentIterator, callback); } @Override @@ -404,18 +417,26 @@ public KeyValue next() { super.next = null; return result; } + + public static WrappedInMemoryWindowStoreIterator emptyIterator() { + return new WrappedInMemoryWindowStoreIterator(null, null, null, it -> { }); + } } - private class WrappedWindowedKeyValueIterator extends InMemoryWindowStoreIteratorWrapper implements KeyValueIterator, byte[]> { + private static class WrappedWindowedKeyValueIterator extends InMemoryWindowStoreIteratorWrapper implements KeyValueIterator, byte[]> { - WrappedWindowedKeyValueIterator() { - super(); - } + private final boolean retainDuplicates; + private final long windowSize; WrappedWindowedKeyValueIterator(final Bytes keyFrom, final Bytes keyTo, - final Iterator>> segmentIterator) { - super(keyFrom, keyTo, segmentIterator); + final Iterator>> segmentIterator, + final ClosingCallback callback, + final boolean retainDuplicates, + final long windowSize) { + super(keyFrom, keyTo, segmentIterator, callback); + this.retainDuplicates = retainDuplicates; + this.windowSize = windowSize; } public Windowed peekNextKey() { diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java index 7df6532a188cb..5c498186811be 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java @@ -412,7 +412,7 @@ public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() { LogCaptureAppender.setClassLoggerToDebug(InMemoryWindowStore.class); final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(); - final KeyValueIterator iterator = store.range(-1, 1); + final KeyValueIterator iterator = store.range(-1, 1); assertFalse(iterator.hasNext()); final List messages = appender.getMessages(); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java index 48c96a28f6c86..8c7325c3cc608 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java @@ -439,7 +439,7 @@ public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() { final Bytes keyFrom = Bytes.wrap(Serdes.Integer().serializer().serialize("", -1)); final Bytes keyTo = Bytes.wrap(Serdes.Integer().serializer().serialize("", 1)); - final KeyValueIterator iterator = cachingStore.findSessions(keyFrom, keyTo, 0L, 10L); + final KeyValueIterator, byte[]> iterator = cachingStore.findSessions(keyFrom, keyTo, 0L, 10L); assertFalse(iterator.hasNext()); final List messages = appender.getMessages(); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java index b0ccc15ca6639..c8d9cc101f5ba 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java @@ -598,7 +598,7 @@ public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() { final Bytes keyFrom = Bytes.wrap(Serdes.Integer().serializer().serialize("", -1)); final Bytes keyTo = Bytes.wrap(Serdes.Integer().serializer().serialize("", 1)); - final KeyValueIterator iterator = cachingStore.fetch(keyFrom, keyTo, 0L, 10L); + final KeyValueIterator, byte[]> iterator = cachingStore.fetch(keyFrom, keyTo, 0L, 10L); assertFalse(iterator.hasNext()); final List messages = appender.getMessages(); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java index df924ec43cb9b..1524d9c6ebda7 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java @@ -605,7 +605,7 @@ public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() { LogCaptureAppender.setClassLoggerToDebug(InMemoryWindowStore.class); final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(); - final KeyValueIterator iterator = windowStore.fetch(-1, 1, 0L, 10L); + final KeyValueIterator, String> iterator = windowStore.fetch(-1, 1, 0L, 10L); assertFalse(iterator.hasNext()); final List messages = appender.getMessages(); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java index 1821913d4771f..80ea4bad89b76 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java @@ -290,7 +290,7 @@ public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() { final String keyFrom = Serdes.String().deserializer().deserialize("", Serdes.Integer().serializer().serialize("", -1)); final String keyTo = Serdes.String().deserializer().deserialize("", Serdes.Integer().serializer().serialize("", 1)); - final KeyValueIterator iterator = sessionStore.findSessions(keyFrom, keyTo, 0L, 10L); + final KeyValueIterator, Long> iterator = sessionStore.findSessions(keyFrom, keyTo, 0L, 10L); assertFalse(iterator.hasNext()); final List messages = appender.getMessages(); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java index 7405e060c580a..33422076e7efb 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java @@ -1426,7 +1426,7 @@ public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() { LogCaptureAppender.setClassLoggerToDebug(InMemoryWindowStore.class); final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(); - final KeyValueIterator iterator = windowStore.fetch(-1, 1, 0L, 10L); + final KeyValueIterator, String> iterator = windowStore.fetch(-1, 1, 0L, 10L); assertFalse(iterator.hasNext()); final List messages = appender.getMessages();