You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2017/10/04 17:14:49 UTC
kafka git commit: KAFKA-5967; Ineffective check of negative value in
CompositeReadOnlyKeyValueStore#approximateNumEntries()
Repository: kafka
Updated Branches:
refs/heads/trunk 5383f9bed -> 5afeddaa9
KAFKA-5967; Ineffective check of negative value in CompositeReadOnlyKeyValueStore#approximateNumEntries()
package name: org.apache.kafka.streams.state.internals
Minor change to approximateNumEntries() method in CompositeReadOnlyKeyValueStore class.
long total = 0;
for (ReadOnlyKeyValueStore<K, V> store : stores) {
total += store.approximateNumEntries();
}
return total < 0 ? Long.MAX_VALUE : total;
The check for negative value seems to account for wrapping. However, wrapping can happen within the for loop. So the check should be performed inside the loop.
Author: siva santhalingam <ss...@netskope.com>
Reviewers: Matthias J. Sax <ma...@confluent.io>, Damian Guy <da...@gmail.com>
Closes #3988 from shivsantham/trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5afeddaa
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5afeddaa
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5afeddaa
Branch: refs/heads/trunk
Commit: 5afeddaa99c48ac827d1cade7812deb83b1f80bd
Parents: 5383f9b
Author: siva santhalingam <ss...@netskope.com>
Authored: Wed Oct 4 10:11:11 2017 -0700
Committer: Damian Guy <da...@gmail.com>
Committed: Wed Oct 4 10:11:11 2017 -0700
----------------------------------------------------------------------
.../CompositeReadOnlyKeyValueStore.java | 5 ++++-
.../CompositeReadOnlyKeyValueStoreTest.java | 19 +++++++++++++++++++
2 files changed, 23 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/5afeddaa/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStore.java
index e3354e4..2c895ef 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStore.java
@@ -104,8 +104,11 @@ public class CompositeReadOnlyKeyValueStore<K, V> implements ReadOnlyKeyValueSto
long total = 0;
for (ReadOnlyKeyValueStore<K, V> store : stores) {
total += store.approximateNumEntries();
+ if (total < 0) {
+ return Long.MAX_VALUE;
+ }
}
- return total < 0 ? Long.MAX_VALUE : total;
+ return total;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/5afeddaa/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
index ad3a1f2..4ff0b90 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
@@ -40,6 +40,7 @@ import static org.junit.Assert.fail;
public class CompositeReadOnlyKeyValueStoreTest {
private final String storeName = "my-store";
+ private final String storeNameA = "my-storeA";
private StateStoreProviderStub stubProviderTwo;
private KeyValueStore<String, String> stubOneUnderlying;
private CompositeReadOnlyKeyValueStore<String, String> theStore;
@@ -257,6 +258,24 @@ public class CompositeReadOnlyKeyValueStoreTest {
assertEquals(Long.MAX_VALUE, theStore.approximateNumEntries());
}
+ @Test
+ public void shouldReturnLongMaxValueOnUnderflow() {
+ stubProviderTwo.addStore(storeName, new NoOpReadOnlyStore<Object, Object>() {
+ @Override
+ public long approximateNumEntries() {
+ return Long.MAX_VALUE;
+ }
+ });
+ stubProviderTwo.addStore(storeNameA, new NoOpReadOnlyStore<Object, Object>() {
+ @Override
+ public long approximateNumEntries() {
+ return Long.MAX_VALUE;
+ }
+ });
+
+ assertEquals(Long.MAX_VALUE, theStore.approximateNumEntries());
+ }
+
private CompositeReadOnlyKeyValueStore<Object, Object> rebalancing() {
return new CompositeReadOnlyKeyValueStore<>(new WrappingStoreProvider(Collections.<StateStoreProvider>singletonList(new StateStoreProviderStub(true))),
QueryableStoreTypes.keyValueStore(), storeName);