KAFKA-5047: NullPointerException while using GlobalKTable in KafkaStreams #2834

dguy · 2017-04-10T16:45:52Z

Skip null keys when initializing GlobalKTables. This is inline with what happens during normal processing.

dguy · 2017-04-10T16:46:06Z

asfbot · 2017-04-10T16:48:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2850/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-10T16:49:04Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2854/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-04-10T16:49:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2849/
Test FAILed (JDK 7 and Scala 2.10).

miguno · 2017-04-10T19:07:30Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java

@@ -174,7 +174,11 @@ private void restoreState(final StateRestoreCallback stateRestoreCallback,
                final ConsumerRecords<byte[], byte[]> records = consumer.poll(100);
                for (ConsumerRecord<byte[], byte[]> record : records) {
                    offset = record.offset() + 1;
-                    stateRestoreCallback.restore(record.key(), record.value());
+                    if (record.key() == null) {
+                        log.warn("null key found at offset %d while restoring state from topic %s and partition %d. Skipping record", record.offset(), record.topic(), record.partition());


Could a log level of WARN wreak havoc in case there's a lot of null keys?

I think the log level we pick here should (a) be consistent with what we use elsewhere (not implying it isn't in this PR!) and (b) it should match the guarantees we have documented for global tables. If, for example, we clearly tell the user that null keys are ignored, we could use a less chatty log level than WARN.

ok, i'll make it debug

actually i'm just going to silently drop it as that is what we do elsewhere

miguno · 2017-04-10T19:09:20Z

...s/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java

+        stateManager.initialize(context);
+        final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+        stateManager.register(store1, false, stateRestoreCallback);
+        assertEquals(1, stateRestoreCallback.restored.size());


Why not assertThat(...)?

because we have to compare KeyValue<byte[], byte[]> and assertThat() won't work., i.e., 2 byte[] with the same contents aren't equal. Need to use assertArrayEquals()

Challenge accepted. ;-)

import org.junit.Test; import static org.assertj.core.api.Assertions.assertThat; public class ArrayComparisonTest { @Test public void shouldCompareByteArrays() throws Exception { byte[] first = "foo".getBytes(); byte[] second = "foo".getBytes(); assertThat(first).isEqualTo(second); } }

This test passes.

AssertJ knows when it is comparing byte arrays.

We don't use that and i'm not adding another dependency

Oh, sorry -- I thought our assertThat was from AssertJ.

But using JUnit's/Hamcrest's assertThat works, too:

import org.junit.Test; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.CoreMatchers.equalTo; public class ArrayComparisonTest { @Test public void shouldCompareByteArrays() throws Exception { byte[] first = "foo".getBytes(); byte[] second = "foo".getBytes(); assertThat(first, equalTo(second)); } }

ok -yeah it is actually the KeyValue#equals() that is causing the problem i was having as it calls byte[].equals underneath which doesn't work. Anyway, i can work around it.

Ok. I didn't want to be picky here, it's just that "someone" (cough) kept telling me in my PRs that we should use assertThat if possible. ;-)

miguno · 2017-04-10T19:13:32Z

...s/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java

+        stateManager.register(store1, false, stateRestoreCallback);
+        assertEquals(1, stateRestoreCallback.restored.size());
+        final KeyValue<byte[], byte[]> restoredKv = stateRestoreCallback.restored.get(0);
+        assertArrayEquals(expectedKey, restoredKv.key);


Why not assertThat?

miguno · 2017-04-10T19:15:13Z

General comment: We should also update the javadocs of GlobalKTable. I'm pretty sure that most users wouldn't expect / know about this.

asfbot · 2017-04-10T19:16:34Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2852/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-10T19:16:53Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2853/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-10T19:51:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2857/
Test PASSed (JDK 8 and Scala 2.11).

enothereska · 2017-04-10T20:23:44Z

@dguy don't we need to do the same check for the normal KTable too?

dguy · 2017-04-11T07:49:50Z

@enothereska For changelogs this will never happen as all key/values that have made it into the changelog will be ok. So restoration is ok. For a KTable that is built via builder.table() then this may happen, now that i think about it. The KTableSource drops null keys, but if it was to be restored it would likely hit the same issue. Thanks!

miguno · 2017-04-11T07:53:14Z

@dguy (/cc @enothereska): I'd also update the KTable javadocs to cover the null-key behavior/contract.

mjsax

LGTM.

Just a general comment: should we add a metric-sensor to report if null-key records get dropped?

mjsax · 2017-04-11T16:19:59Z

streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java

@@ -54,6 +54,8 @@
 *}</pre>
 * Note that in contrast to {@link KTable} a {@code GlobalKTable}'s state holds a full copy of the underlying topic,
 * thus all keys can be queried locally.
+ * <p>
+ * Records from the source topic that have null keys are dropped


Nit. . missing at the end.

mjsax · 2017-04-11T16:25:12Z

streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java

@@ -51,6 +51,8 @@
 *     ReadOnlyKeyValueStore view = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
 *     view.get(key);
 *}</pre>
+ *<p>
+ * Records from the source topic that have null keys are dropped


dguy · 2017-04-12T08:03:39Z

@mjsax that is a good idea, though i i'll file a JIRA for it as i think it applies more broadly than this particular fix. We drop records in various places and don't record it.

https://issues.apache.org/jira/browse/KAFKA-5058

asfbot · 2017-04-12T08:57:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2906/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-04-12T09:03:30Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2902/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-12T09:04:23Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2901/
Test PASSed (JDK 7 and Scala 2.10).

dguy · 2017-04-12T13:42:29Z

@enothereska i updated it to check KTable. Can you please make another pass? @ijuma would you mind having a look once @enothereska has given the thumbs up?

enothereska · 2017-04-13T18:07:51Z

LGTM thanks.

dguy · 2017-04-20T11:09:17Z

@guozhangwang can you please merge?

guozhangwang · 2017-04-20T21:27:00Z

LGTM and Merged to trunk.

@dguy I tried to cherry-pick to 0.10.2 but the conflicts are quite complex for me to resolve-forward, could you file a separate PR for 0.10.2 also?

skip null keys when restoring global state

ada1371

fix checkstyle

1d1137c

miguno reviewed Apr 10, 2017

View reviewed changes

dguy added 2 commits April 11, 2017 09:20

check for null when restoring KTable. remove logging

a06dd39

use assertThat to keep michael happy ;-)

ea8fe48

mjsax reviewed Apr 11, 2017

View reviewed changes

address comments

257fbc2

asfgit closed this in 3c471d2 Apr 20, 2017

dguy deleted the kafka-5047 branch May 16, 2017 14:05

KAFKA-5047: NullPointerException while using GlobalKTable in KafkaStreams #2834

KAFKA-5047: NullPointerException while using GlobalKTable in KafkaStreams #2834

Conversation

dguy commented Apr 10, 2017

dguy commented Apr 10, 2017

asfbot commented Apr 10, 2017

asfbot commented Apr 10, 2017

asfbot commented Apr 10, 2017

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

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

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

miguno commented Apr 10, 2017

asfbot commented Apr 10, 2017

asfbot commented Apr 10, 2017

asfbot commented Apr 10, 2017

enothereska commented Apr 10, 2017

dguy commented Apr 11, 2017

miguno commented Apr 11, 2017

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

dguy commented Apr 12, 2017 • edited Loading

asfbot commented Apr 12, 2017

asfbot commented Apr 12, 2017

asfbot commented Apr 12, 2017

dguy commented Apr 12, 2017

enothereska commented Apr 13, 2017

dguy commented Apr 20, 2017

guozhangwang commented Apr 20, 2017

dguy commented Apr 12, 2017 •

edited

Loading