forked from apache/kafka
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
KAFKA-4863; Querying window store may return unwanted keys
Make sure that the iterator returned from `WindowStore.fetch(..)` only returns matching keys, rather than all keys that are a prefix match. Author: Damian Guy <[email protected]> Reviewers: Eno Thereska, Guozhang Wang Closes apache#2662 from dguy/kafka-4863
- Loading branch information
1 parent
d3b8ff0
commit 9e4548d
Showing
7 changed files
with
663 additions
and
482 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
73 changes: 73 additions & 0 deletions
73
streams/src/main/java/org/apache/kafka/streams/state/internals/FilteredCacheIterator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.kafka.streams.state.internals; | ||
|
|
||
| import org.apache.kafka.common.utils.Bytes; | ||
| import org.apache.kafka.streams.KeyValue; | ||
|
|
||
| import java.util.NoSuchElementException; | ||
|
|
||
| class FilteredCacheIterator implements PeekingKeyValueIterator<Bytes, LRUCacheEntry> { | ||
| private final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator; | ||
| private final HasNextCondition hasNextCondition; | ||
|
|
||
| FilteredCacheIterator(final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator, | ||
| final HasNextCondition hasNextCondition) { | ||
| this.cacheIterator = cacheIterator; | ||
| this.hasNextCondition = hasNextCondition; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| // no-op | ||
| } | ||
|
|
||
| @Override | ||
| public Bytes peekNextKey() { | ||
| if (!hasNext()) { | ||
| throw new NoSuchElementException(); | ||
| } | ||
| return cacheIterator.peekNextKey(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean hasNext() { | ||
| return hasNextCondition.hasNext(cacheIterator); | ||
| } | ||
|
|
||
| @Override | ||
| public KeyValue<Bytes, LRUCacheEntry> next() { | ||
| if (!hasNext()) { | ||
| throw new NoSuchElementException(); | ||
| } | ||
| return cacheIterator.next(); | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void remove() { | ||
| throw new UnsupportedOperationException(); | ||
| } | ||
|
|
||
| @Override | ||
| public KeyValue<Bytes, LRUCacheEntry> peekNext() { | ||
| if (!hasNext()) { | ||
| throw new NoSuchElementException(); | ||
| } | ||
| return cacheIterator.peekNext(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
118 changes: 118 additions & 0 deletions
118
...ams/src/test/java/org/apache/kafka/streams/state/internals/FilteredCacheIteratorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.kafka.streams.state.internals; | ||
|
|
||
| import org.apache.kafka.common.utils.Bytes; | ||
| import org.apache.kafka.common.utils.Utils; | ||
| import org.apache.kafka.streams.KeyValue; | ||
| import org.apache.kafka.streams.state.KeyValueIterator; | ||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| import static org.apache.kafka.test.StreamsTestUtils.toList; | ||
| import static org.hamcrest.CoreMatchers.equalTo; | ||
| import static org.hamcrest.MatcherAssert.assertThat; | ||
| import static org.junit.Assert.assertFalse; | ||
| import static org.junit.Assert.assertTrue; | ||
|
|
||
| public class FilteredCacheIteratorTest { | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| private final InMemoryKeyValueStore<Bytes, LRUCacheEntry> store = new InMemoryKeyValueStore("name", null, null); | ||
| private final KeyValue<Bytes, LRUCacheEntry> firstEntry = KeyValue.pair(Bytes.wrap("a".getBytes()), | ||
| new LRUCacheEntry("1".getBytes())); | ||
| private final List<KeyValue<Bytes, LRUCacheEntry>> entries = Utils.mkList( | ||
| firstEntry, | ||
| KeyValue.pair(Bytes.wrap("b".getBytes()), | ||
| new LRUCacheEntry("2".getBytes())), | ||
| KeyValue.pair(Bytes.wrap("c".getBytes()), | ||
| new LRUCacheEntry("3".getBytes()))); | ||
|
|
||
| private FilteredCacheIterator allIterator; | ||
| private FilteredCacheIterator firstEntryIterator; | ||
|
|
||
| @Before | ||
| public void before() { | ||
| store.putAll(entries); | ||
| final HasNextCondition allCondition = new HasNextCondition() { | ||
| @Override | ||
| public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { | ||
| return iterator.hasNext(); | ||
| } | ||
| }; | ||
| allIterator = new FilteredCacheIterator( | ||
| new DelegatingPeekingKeyValueIterator<>("", | ||
| store.all()), allCondition); | ||
|
|
||
| final HasNextCondition firstEntryCondition = new HasNextCondition() { | ||
| @Override | ||
| public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { | ||
| return iterator.hasNext() && iterator.peekNextKey().equals(firstEntry.key); | ||
| } | ||
| }; | ||
| firstEntryIterator = new FilteredCacheIterator( | ||
| new DelegatingPeekingKeyValueIterator<>("", | ||
| store.all()), firstEntryCondition); | ||
|
|
||
| } | ||
|
|
||
| @Test | ||
| public void shouldAllowEntryMatchingHasNextCondition() throws Exception { | ||
| final List<KeyValue<Bytes, LRUCacheEntry>> keyValues = toList(allIterator); | ||
| assertThat(keyValues, equalTo(entries)); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldPeekNextKey() throws Exception { | ||
| while (allIterator.hasNext()) { | ||
| final Bytes nextKey = allIterator.peekNextKey(); | ||
| final KeyValue<Bytes, LRUCacheEntry> next = allIterator.next(); | ||
| assertThat(next.key, equalTo(nextKey)); | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldPeekNext() throws Exception { | ||
| while (allIterator.hasNext()) { | ||
| final KeyValue<Bytes, LRUCacheEntry> peeked = allIterator.peekNext(); | ||
| final KeyValue<Bytes, LRUCacheEntry> next = allIterator.next(); | ||
| assertThat(peeked, equalTo(next)); | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldNotHaveNextIfHasNextConditionNotMet() throws Exception { | ||
| assertTrue(firstEntryIterator.hasNext()); | ||
| firstEntryIterator.next(); | ||
| assertFalse(firstEntryIterator.hasNext()); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldFilterEntriesNotMatchingHasNextCondition() throws Exception { | ||
| final List<KeyValue<Bytes, LRUCacheEntry>> keyValues = toList(firstEntryIterator); | ||
| assertThat(keyValues, equalTo(Utils.mkList(firstEntry))); | ||
| } | ||
|
|
||
| @Test(expected = UnsupportedOperationException.class) | ||
| public void shouldThrowUnsupportedOperationExeceptionOnRemove() throws Exception { | ||
| allIterator.remove(); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.