From 3d54963d3e95a7ff1322c178263b9431bf7e615b Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Wed, 17 Apr 2019 15:48:13 -0700 Subject: [PATCH 1/6] Removed unnecessary this and initialized data structures in declaration --- .../internals/InMemoryKeyValueStore.java | 28 +++++---- .../state/internals/InMemoryWindowStore.java | 57 +++++++++---------- 2 files changed, 40 insertions(+), 45 deletions(-) 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..c2e9d6150bba4 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,14 +78,11 @@ 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 @@ -113,7 +110,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 +122,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 +146,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 +164,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 new WrappedInMemoryWindowStoreIterator( - key, key, this.segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + key, key, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @Deprecated @@ -193,14 +190,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 new WrappedWindowedKeyValueIterator( - from, to, this.segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + from, to, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @Deprecated @@ -209,24 +206,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 new WrappedWindowedKeyValueIterator( - null, null, this.segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + 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()); + null, null, segmentMap.tailMap(minTime, false).entrySet().iterator()); } @Override @@ -236,7 +233,7 @@ public boolean persistent() { @Override public boolean isOpen() { - return this.open; + return open; } @Override @@ -246,16 +243,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() { @@ -292,14 +289,14 @@ private abstract class InMemoryWindowStoreIteratorWrapper implements Comparable< // Default constructor sets up a dummy iterator when no results are returned (eg entire fetch range is expired) InMemoryWindowStoreIteratorWrapper() { - this.allKeys = false; + allKeys = false; recordIterator = null; } InMemoryWindowStoreIteratorWrapper(final Bytes keyFrom, final Bytes keyTo, final Iterator>> segmentIterator) { - this.allKeys = (keyFrom == null) && (keyTo == null); + allKeys = (keyFrom == null) && (keyTo == null); if (retainDuplicates && !allKeys) { this.keyFrom = wrapForDups(keyFrom, 0); this.keyTo = wrapForDups(keyTo, Integer.MAX_VALUE); @@ -309,7 +306,7 @@ private abstract class InMemoryWindowStoreIteratorWrapper implements Comparable< } this.segmentIterator = segmentIterator; - this.recordIterator = setRecordIterator(); + recordIterator = setRecordIterator(); openIterators.add(this); } From 664404c078c1456f23cac9f66691bb9e175c7032 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Wed, 17 Apr 2019 15:54:41 -0700 Subject: [PATCH 2/6] Added missing types to KeyValueIterators in store tests --- .../kafka/streams/state/internals/InMemoryWindowStore.java | 7 +------ .../streams/state/internals/AbstractKeyValueStoreTest.java | 2 +- .../streams/state/internals/CachingSessionStoreTest.java | 2 +- .../streams/state/internals/CachingWindowStoreTest.java | 2 +- .../streams/state/internals/InMemoryWindowStoreTest.java | 2 +- .../streams/state/internals/RocksDBSessionStoreTest.java | 2 +- .../streams/state/internals/RocksDBWindowStoreTest.java | 2 +- 7 files changed, 7 insertions(+), 12 deletions(-) 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 c2e9d6150bba4..b136caf41dd5b 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 @@ -86,7 +86,6 @@ public String name() { } @Override - @SuppressWarnings("unchecked") public void init(final ProcessorContext context, final StateStore root) { this.context = (InternalProcessorContext) context; @@ -276,7 +275,7 @@ private static Bytes getKey(final Bytes keyBytes) { } - private abstract class InMemoryWindowStoreIteratorWrapper implements Comparable { + private abstract class InMemoryWindowStoreIteratorWrapper { private Iterator>> segmentIterator; private Iterator> recordIterator; @@ -365,10 +364,6 @@ Iterator> setRecordIterator() { Long minTime() { return currentTime; } - - public int compareTo(final InMemoryWindowStoreIteratorWrapper other) { - return (int) (minTime() - other.minTime()); - } } private class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIteratorWrapper implements WindowStoreIterator { 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(); From af5e829ee7a4708e37c819fea2c84cf59bfe6c13 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Wed, 17 Apr 2019 16:33:55 -0700 Subject: [PATCH 3/6] Safely register new iterators, iterator classes should be static --- .../state/internals/InMemoryWindowStore.java | 96 ++++++++++++++----- 1 file changed, 71 insertions(+), 25 deletions(-) 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 b136caf41dd5b..a61b301f99638 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 @@ -169,8 +169,8 @@ public WindowStoreIterator fetch(final Bytes key, final long timeFrom, f return new WrappedInMemoryWindowStoreIterator(); } - return new WrappedInMemoryWindowStoreIterator( - key, key, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); + return registerNewWindowStoreIterator( + key, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @Deprecated @@ -195,7 +195,7 @@ public KeyValueIterator, byte[]> fetch(final Bytes from, return new WrappedWindowedKeyValueIterator(); } - return new WrappedWindowedKeyValueIterator( + return registerNewWindowedKeyValueIterator( from, to, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @@ -211,7 +211,7 @@ public KeyValueIterator, byte[]> fetchAll(final long timeFrom, f return new WrappedWindowedKeyValueIterator(); } - return new WrappedWindowedKeyValueIterator( + return registerNewWindowedKeyValueIterator( null, null, segmentMap.subMap(minTime, true, timeTo, true).entrySet().iterator()); } @@ -221,7 +221,7 @@ public KeyValueIterator, byte[]> all() { final long minTime = observedStreamTime - retentionPeriod; - return new WrappedWindowedKeyValueIterator( + return registerNewWindowedKeyValueIterator( null, null, segmentMap.tailMap(minTime, false).entrySet().iterator()); } @@ -275,7 +275,43 @@ private static Bytes getKey(final Bytes keyBytes) { } - private abstract class InMemoryWindowStoreIteratorWrapper { + 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(Bytes keyFrom, + Bytes keyTo, + final Iterator>> segmentIterator) { + if (retainDuplicates && !(keyFrom == null || keyTo == null)) { + keyFrom = wrapForDups(keyFrom, 0); + keyTo = wrapForDups(keyTo, Integer.MAX_VALUE); + } + + final WrappedWindowedKeyValueIterator iterator = + new WrappedWindowedKeyValueIterator(keyFrom, + keyTo, + 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; @@ -283,31 +319,30 @@ private abstract class InMemoryWindowStoreIteratorWrapper { private long currentTime; private final boolean allKeys; - private Bytes keyFrom; - private Bytes keyTo; + private final Bytes keyFrom; + private final Bytes keyTo; + private final ClosingCallback callback; // Default constructor sets up a dummy iterator when no results are returned (eg entire fetch range is expired) InMemoryWindowStoreIteratorWrapper() { allKeys = false; recordIterator = null; + keyFrom = null; + keyTo = null; + callback = null; } InMemoryWindowStoreIteratorWrapper(final Bytes keyFrom, final Bytes keyTo, - final Iterator>> segmentIterator) { + final Iterator>> segmentIterator, + final ClosingCallback callback) { + this.keyFrom = keyFrom; + this.keyTo = keyTo; 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; - } this.segmentIterator = segmentIterator; + this.callback = callback; recordIterator = setRecordIterator(); - - openIterators.add(this); } public boolean hasNext() { @@ -328,7 +363,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 @@ -366,7 +401,7 @@ Long minTime() { } } - private class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIteratorWrapper implements WindowStoreIterator { + private static class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIteratorWrapper implements WindowStoreIterator { WrappedInMemoryWindowStoreIterator() { super(); @@ -374,8 +409,9 @@ private class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIter 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 @@ -398,16 +434,26 @@ public KeyValue next() { } } - private class WrappedWindowedKeyValueIterator extends InMemoryWindowStoreIteratorWrapper implements KeyValueIterator, byte[]> { + private static class WrappedWindowedKeyValueIterator extends InMemoryWindowStoreIteratorWrapper implements KeyValueIterator, byte[]> { + + private final boolean retainDuplicates; + private final long windowSize; WrappedWindowedKeyValueIterator() { super(); + this.windowSize = 0; + this.retainDuplicates = false; } 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() { From cee2e5d0ebc688e419c0453ee366050af73aef6c Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Thu, 18 Apr 2019 13:00:36 -0700 Subject: [PATCH 4/6] Don't modify keyFrom/keyTo --- .../state/internals/InMemoryWindowStore.java | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) 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 a61b301f99638..001d9fe11baaf 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 @@ -287,17 +287,15 @@ private WrappedInMemoryWindowStoreIterator registerNewWindowStoreIterator(final return iterator; } - private WrappedWindowedKeyValueIterator registerNewWindowedKeyValueIterator(Bytes keyFrom, - Bytes keyTo, - final Iterator>> segmentIterator) { - if (retainDuplicates && !(keyFrom == null || keyTo == null)) { - keyFrom = wrapForDups(keyFrom, 0); - keyTo = wrapForDups(keyTo, Integer.MAX_VALUE); - } + 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(keyFrom, - keyTo, + new WrappedWindowedKeyValueIterator(from, + to, segmentIterator, openIterators::remove, retainDuplicates, From a41ba64615c0a4012ff2b76a0d7d6a0097e22769 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Thu, 18 Apr 2019 14:23:30 -0700 Subject: [PATCH 5/6] Use static emptyIterator() method instead of dummy constructor --- .../state/internals/InMemoryWindowStore.java | 31 +++++-------------- 1 file changed, 8 insertions(+), 23 deletions(-) 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 001d9fe11baaf..0f2e3b68031d4 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 @@ -166,7 +166,7 @@ public WindowStoreIterator fetch(final Bytes key, final long timeFrom, f final long minTime = Math.max(timeFrom, observedStreamTime - retentionPeriod + 1); if (timeTo < minTime) { - return new WrappedInMemoryWindowStoreIterator(); + return WrappedInMemoryWindowStoreIterator.emptyIterator(); } return registerNewWindowStoreIterator( @@ -192,7 +192,7 @@ public KeyValueIterator, byte[]> fetch(final Bytes from, final long minTime = Math.max(timeFrom, observedStreamTime - retentionPeriod + 1); if (timeTo < minTime) { - return new WrappedWindowedKeyValueIterator(); + return KeyValueIterators.emptyIterator(); } return registerNewWindowedKeyValueIterator( @@ -208,7 +208,7 @@ public KeyValueIterator, byte[]> fetchAll(final long timeFrom, f final long minTime = Math.max(timeFrom, observedStreamTime - retentionPeriod + 1); if (timeTo < minTime) { - return new WrappedWindowedKeyValueIterator(); + return KeyValueIterators.emptyIterator(); } return registerNewWindowedKeyValueIterator( @@ -321,15 +321,6 @@ private static abstract class InMemoryWindowStoreIteratorWrapper { private final Bytes keyTo; private final ClosingCallback callback; - // Default constructor sets up a dummy iterator when no results are returned (eg entire fetch range is expired) - InMemoryWindowStoreIteratorWrapper() { - allKeys = false; - recordIterator = null; - keyFrom = null; - keyTo = null; - callback = null; - } - InMemoryWindowStoreIteratorWrapper(final Bytes keyFrom, final Bytes keyTo, final Iterator>> segmentIterator, @@ -340,7 +331,7 @@ private static abstract class InMemoryWindowStoreIteratorWrapper { this.segmentIterator = segmentIterator; this.callback = callback; - recordIterator = setRecordIterator(); + recordIterator = segmentIterator == null ? null : setRecordIterator(); } public boolean hasNext() { @@ -401,10 +392,6 @@ Long minTime() { private static class WrappedInMemoryWindowStoreIterator extends InMemoryWindowStoreIteratorWrapper implements WindowStoreIterator { - WrappedInMemoryWindowStoreIterator() { - super(); - } - WrappedInMemoryWindowStoreIterator(final Bytes keyFrom, final Bytes keyTo, final Iterator>> segmentIterator, @@ -430,6 +417,10 @@ public KeyValue next() { super.next = null; return result; } + + public static WrappedInMemoryWindowStoreIterator emptyIterator() { + return new WrappedInMemoryWindowStoreIterator(null, null, null, it -> {}); + } } private static class WrappedWindowedKeyValueIterator extends InMemoryWindowStoreIteratorWrapper implements KeyValueIterator, byte[]> { @@ -437,12 +428,6 @@ private static class WrappedWindowedKeyValueIterator extends InMemoryWindowStore private final boolean retainDuplicates; private final long windowSize; - WrappedWindowedKeyValueIterator() { - super(); - this.windowSize = 0; - this.retainDuplicates = false; - } - WrappedWindowedKeyValueIterator(final Bytes keyFrom, final Bytes keyTo, final Iterator>> segmentIterator, From fcd32003705334c5ed43a92b206f9df2f355de64 Mon Sep 17 00:00:00 2001 From: ableegoldman Date: Thu, 18 Apr 2019 17:24:31 -0700 Subject: [PATCH 6/6] Fixed checkstyle --- .../kafka/streams/state/internals/InMemoryWindowStore.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 0f2e3b68031d4..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 @@ -419,7 +419,7 @@ public KeyValue next() { } public static WrappedInMemoryWindowStoreIterator emptyIterator() { - return new WrappedInMemoryWindowStoreIterator(null, null, null, it -> {}); + return new WrappedInMemoryWindowStoreIterator(null, null, null, it -> { }); } }