KAFKA-12396 added a nullcheck before trying to retrieve a key #10548

Nathan22177 · 2021-04-16T12:44:23Z

No description provided.

Vasilisck · 2021-04-16T12:44:59Z

LGTM

mjsax · 2021-04-16T17:18:17Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

@@ -326,6 +326,7 @@ public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {

    @Override
    public synchronized byte[] get(final Bytes key) {
+        Objects.requireNonNull(key, "key cannot be null");


I think we should add this to MeteredKeyValueStore instead -- this way, we get the same guard for the in-memory key-value store (or other custom stores).

We should also add this check to other methods that accept a key, like put, putIfAbsent etc...

Furthermore, we should also add it for other store types, ie, MeteredTimestampedKeyValueStore, MeteredWindowedStore, MeteredTimestampedWindowStore and MeteredSessionStore.

Last, but not least, we should add corresponding unit tests.

This check can be remove, as we do the check in the "metered" store that will wrap this one.

…test coverage for aforementioned nullchecks

mjsax

A few methods are missing:

MeteredKeyValueStore:

putAll() (we should iterator over all entries to check if key might be null
prefixScan() (prefix should not be null either I guess? \cc @guozhangwang @cadonna
range() (both keys)
reverseRange (both keys)

MeteredSessionStore:

put() (should verify sessionKey != null and sessionKey.key() != null)
remove()
fetchSession()
fetch(K)
backwardFetch(K)
fetch(K, K)
backwardFetch(K, K)
findSession(K, long, long)
backwardFindSession
findSession(K, K, long, long)
backwardFindSession(K, K, long, long)

Thanks for adding tests!

mjsax · 2021-04-21T01:15:02Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

@@ -290,6 +295,7 @@ protected V outerValue(final byte[] value) {
    }

    protected Bytes keyBytes(final K key) {
+        Objects.requireNonNull(key, "key cannot be null");


This method is not public so no need to add this check

mjsax · 2021-04-21T01:19:47Z

.../src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java

@@ -59,6 +61,7 @@


    public RawAndDeserializedValue<V> getWithBinary(final K key) {
+        Objects.requireNonNull(key, "key cannot be null");


Actually not public API -- don't need this check.

mjsax · 2021-04-21T01:19:51Z

.../src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java

@@ -73,6 +76,7 @@
    public boolean putIfDifferentValues(final K key,
                                        final ValueAndTimestamp<V> newValue,
                                        final byte[] oldSerializedValue) {
+        Objects.requireNonNull(key, "key cannot be null");


Actually not public API -- don't need this check.

mjsax · 2021-04-21T01:21:57Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

@@ -326,6 +326,7 @@ public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {

    @Override
    public synchronized byte[] get(final Bytes key) {
+        Objects.requireNonNull(key, "key cannot be null");


This check can be remove, as we do the check in the "metered" store that will wrap this one.

…MeteredTimestampedKeyValueStore, added new checks to MeteredKeyValueStore, covered new checks and old checks for MeteredSessionStoreTest

Nathan22177 · 2021-04-21T15:13:43Z

@mjsax

MeteredSessionStore:

put() (should verify sessionKey != null and sessionKey.key() != null)

remove()

fetchSession()

fetch(K)

backwardFetch(K)

fetch(K, K)

backwardFetch(K, K)

findSession(K, long, long)

backwardFindSession

findSession(K, K, long, long)

backwardFindSession(K, K, long, long)

All of them already had the checks. Some lacked test coverage, though. I made changes according to your comments and added some tests.

mjsax · 2021-04-21T20:07:55Z

All of them already had the checks.

Sweet. I did not double check the code before.

Seems, put(final Windowed<K> sessionKey,...) and remove(final Windowed<K> sessionKey) don't check for sessionKey.key() != null though? Can we add this check?

cadonna · 2021-04-22T08:54:00Z

* prefixScan() (prefix should not be `null` either I guess? \cc @guozhangwang @cadonna

Yes! Actually, the implementations in RocksDBStore and InMemoryKeyValueStore already have a null check.

   public <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan(final P prefix,
                                                                                    final PS prefixKeySerializer) {
        Objects.requireNonNull(prefix, "prefix cannot be null");
        Objects.requireNonNull(prefixKeySerializer, "prefixKeySerializer cannot be null");

I think it would make sense to move them to the MeteredKeyValueStore.

Nathan22177 · 2021-04-22T13:23:01Z

Seems, put(final Windowed<K> sessionKey,...) and remove(final Windowed<K> sessionKey) don't check for sessionKey.key() != null though? Can we add this check?

They do? Since 2017 according to annotations? Might be that we're talking different classes?

Nathan22177 · 2021-04-22T13:25:41Z

And both are covered with tests.

Nathan22177 · 2021-04-22T13:29:41Z

Tests for aforementioned checks were added here:
https://github.com/apache/kafka/pull/9520/files#diff-dc416cbdf3efca7ee284fe1e27110e737c4178b7a1cc0696d9d7a72bb8e6e764

mjsax · 2021-04-22T18:05:22Z

They do? Since 2017 according to annotations? Might be that we're talking different classes?

The input parameter is Windowed<K> sessionKey and the check is if sessionKey != null -- however, sessionKey wraps a key object, and my ask was to also check if the wrapped key is not null, ie, to add sessionKey.key() != null check (note the .get() call that extract the wrapped key from sessionKey parameter). Sorry for the confusion.

Nathan22177 · 2021-04-22T20:23:27Z

Ooohhhh. I see. Yea, sorry, I also completely overlooked the wrapped key part. I'll fix it.

mjsax

Thanks for staying on top of this! Seems we find more corner case while reviewing... A few last comments.

mjsax · 2021-04-22T21:49:26Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

@@ -219,11 +224,19 @@ public V putIfAbsent(final K key,

    @Override
    public void putAll(final List<KeyValue<K, V>> entries) {
+        final List<KeyValue<K, V>> possiblyNullKeys = entries


I think we could simplify this to a one liner?

entries.forEach(entry -> Objects.requireNonNull(entry.key, "key cannot be null"));

mjsax · 2021-04-22T21:50:39Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

@@ -234,13 +247,15 @@ public V delete(final K key) {

    @Override
    public <PS extends Serializer<P>, P> KeyValueIterator<K, V> prefixScan(final P prefix, final PS prefixKeySerializer) {
-
+        Objects.requireNonNull(prefix, "key cannot be null");


As mentioned by @cadonna the wrapped stores, also check prefixKeySerializer for null -- thus might be good to move both check here.

I think we can also remove both checks in RocksDBStore and InMemoryKeyValueStore -- they seems to be redundant now?

I think we can also remove both checks in RocksDBStore and InMemoryKeyValueStore -- they seem to be redundant now?

they are both different implementations, aren't they?

I don't understand how they will be checked if we only leave it in MeteredKeyValueStore.

I wrote a quick test to see if it still throws NPE in InMemoryKeyValueStore without the check - it did, I am confused, but they are, indeed, redundant.
I'll leave the tests in both RocksDBStoreTest and InMemoryKeyValueStore bc why not.

KafkaStreams runtime always "wraps" any store with a corresponding MeteredXxxStore (cf https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilder.java#L42-L49) -- those MeteredXxxStores do the transaction from objects to bytes (ie they use the serdes) and also track state store metrics. (Note that stores provided to the runtime always have type <Bytes, byte[]> while they are exposed to Processors as <K,V> types.)

Thus, when you call context.stateStore(...) you always get a MeteredXxxStore object -- of course, those details are hidden behind the interface type.

This architecture allows us to unify code and separate concerns. In fact, it also allows us to add/remove more "layers": we can insert a "caching layer" (cf. https://kafka.apache.org/28/documentation/streams/developer-guide/memory-mgmt.html) and a "change logging layer" (both are inserted by default).

Oooooh, I see.

mjsax · 2021-04-22T21:53:19Z

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java

@@ -472,11 +472,26 @@ public void shouldThrowNullPointerOnRemoveIfKeyIsNull() {
        assertThrows(NullPointerException.class, () -> store.remove(null));
    }

+    @Test
+    public void shouldThrowNullPointerOnPutIfWrappedKeyIsNull() {
+        assertThrows(NullPointerException.class, () -> store.put(new Windowed<>(null, new SessionWindow(0, 0)), "a"));


This test remind me, that the SessionWindow what is wrapped should not be null either.

…toring in MeteredKeyValueStore\InMemoryKeyValueStore\RocksDBStoreTest

mjsax · 2021-04-25T23:51:54Z

Seems there is some checkstyle error:

[2021-04-22T23:02:56.789Z] [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-10548/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:35:8: Unused import - java.util.Objects. [UnusedImports]

mjsax · 2021-04-26T16:56:44Z

streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java

@@ -402,6 +402,11 @@ public void shouldReturnKeysWithGivenPrefix() {
        assertThat(valuesWithPrefix.get(2), is("b"));
    }

+    @Test
+    public void shouldThrowNullPointerIfPrefixKeySerializerIsNull() {


This test fails now, as the check was removed. No need to add this test any longer in this class.

mjsax · 2021-04-26T16:59:20Z

To avoid checkstyle and test issues, and reduce review turn-around time, I would recommend to run unit tests and checkstyle locally before pushing update: ./gradlew clean stream:unitTest streams:checkstyleMain streams:checkstyleTest

Nathan22177 · 2021-04-29T22:25:59Z

To avoid checkstyle and test issues, and reduce review turn-around time, I would recommend to run unit tests and checkstyle locally before pushing update: ./gradlew clean stream:unitTest streams:checkstyleMain streams:checkstyleTest

Yeah, sorry, I forget to do that sometimes.

mjsax · 2021-04-30T02:47:41Z

Thanks for the PR @Nathan22177! Merged to trunk.

KAFKA-12396 added a nullcheck before trying to retrieve a key

26395a5

mjsax added the streams label Apr 16, 2021

mjsax reviewed Apr 16, 2021

View reviewed changes

KAFKA-12396 elevated nullcheck to store types and provided some unit …

b06e7ae

…test coverage for aforementioned nullchecks

mjsax reviewed Apr 21, 2021

View reviewed changes

KAFKA-12396 removed unnecessary checks from MeteredKeyValueStore and …

8091bf4

…MeteredTimestampedKeyValueStore, added new checks to MeteredKeyValueStore, covered new checks and old checks for MeteredSessionStoreTest

KAFKA-12396 got rid of the null check in RocksDBStore

c6703a0

KAFKA-12396 reverted RockDBStore to it's original state

983cc95

KAFKA-12396 checkstyle errors fixed

7142aa2

mjsax reviewed Apr 22, 2021

View reviewed changes

nathan22177 added 2 commits April 23, 2021 01:50

KAFKA-12396 new checks\tests for MeteredSessionStoreTest, small refac…

c9aef82

…toring in MeteredKeyValueStore\InMemoryKeyValueStore\RocksDBStoreTest

KAFKA-12396 cosmetic changes in tests signatures

5e61459

KAFKA-12396 removed unused import

3d6bc9b

mjsax reviewed Apr 26, 2021

View reviewed changes

KAFKA-12396 removed obsolete test

12038c0

mjsax merged commit e454bec into apache:trunk Apr 30, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-12396 added a nullcheck before trying to retrieve a key #10548

KAFKA-12396 added a nullcheck before trying to retrieve a key #10548

Nathan22177 commented Apr 16, 2021

Vasilisck commented Apr 16, 2021

mjsax Apr 16, 2021

Nathan22177 Apr 16, 2021

mjsax Apr 21, 2021

mjsax left a comment

mjsax Apr 21, 2021

mjsax Apr 21, 2021

mjsax Apr 21, 2021

mjsax Apr 21, 2021

Nathan22177 commented Apr 21, 2021

mjsax commented Apr 21, 2021

cadonna commented Apr 22, 2021

Nathan22177 commented Apr 22, 2021 •

edited

Nathan22177 commented Apr 22, 2021

Nathan22177 commented Apr 22, 2021

mjsax commented Apr 22, 2021

Nathan22177 commented Apr 22, 2021 •

edited

mjsax left a comment

mjsax Apr 22, 2021

mjsax Apr 22, 2021

Nathan22177 Apr 22, 2021

Nathan22177 Apr 22, 2021

mjsax Apr 25, 2021 •

edited

Nathan22177 Apr 27, 2021

mjsax Apr 22, 2021

mjsax commented Apr 25, 2021

mjsax Apr 26, 2021

mjsax commented Apr 26, 2021

Nathan22177 commented Apr 29, 2021

mjsax commented Apr 30, 2021

		@@ -59,6 +61,7 @@


		public RawAndDeserializedValue<V> getWithBinary(final K key) {
		Objects.requireNonNull(key, "key cannot be null");

KAFKA-12396 added a nullcheck before trying to retrieve a key #10548

KAFKA-12396 added a nullcheck before trying to retrieve a key #10548

Conversation

Nathan22177 commented Apr 16, 2021

Vasilisck commented Apr 16, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Nathan22177 commented Apr 21, 2021

mjsax commented Apr 21, 2021

cadonna commented Apr 22, 2021

Nathan22177 commented Apr 22, 2021 • edited

Nathan22177 commented Apr 22, 2021

Nathan22177 commented Apr 22, 2021

mjsax commented Apr 22, 2021

Nathan22177 commented Apr 22, 2021 • edited

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Apr 25, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Apr 25, 2021

Choose a reason for hiding this comment

mjsax commented Apr 26, 2021

Nathan22177 commented Apr 29, 2021

mjsax commented Apr 30, 2021

Nathan22177 commented Apr 22, 2021 •

edited

Nathan22177 commented Apr 22, 2021 •

edited

mjsax Apr 25, 2021 •

edited