Repository: kafka Updated Branches: refs/heads/0.10.0 175fbb559 -> 84c995d8d
MINOR: Check null keys in KTableSource Author: Guozhang Wang <[email protected]> Reviewers: Damian Guy, Matthias J. Sax Closes #1521 from guozhangwang/Kminor-check-nullkey-ktable-source (cherry picked from commit 91135ea33a7231e9b17cfc74c78dc5223475834b) Signed-off-by: Guozhang Wang <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/84c995d8 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/84c995d8 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/84c995d8 Branch: refs/heads/0.10.0 Commit: 84c995d8d55cb920c0fb139107fb08c766d9476e Parents: 175fbb5 Author: Guozhang Wang <[email protected]> Authored: Sat Jun 18 12:02:17 2016 -0700 Committer: Guozhang Wang <[email protected]> Committed: Sat Jun 18 12:02:25 2016 -0700 ---------------------------------------------------------------------- .../org/apache/kafka/streams/kstream/KStreamBuilder.java | 2 ++ .../kafka/streams/kstream/internals/KTableSource.java | 10 +++++++++- 2 files changed, 11 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/84c995d8/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java index 9d90ba0..9f7f9c4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java @@ -77,6 +77,7 @@ public class KStreamBuilder extends TopologyBuilder { /** * Create a {@link KTable} instance for the specified topic. + * Record keys of the topic should never by null, otherwise an exception will be thrown at runtime. * The default deserializers specified in the config are used. * * @param topic the topic name; cannot be null @@ -88,6 +89,7 @@ public class KStreamBuilder extends TopologyBuilder { /** * Create a {@link KTable} instance for the specified topic. + * Record keys of the topic should never by null, otherwise an exception will be thrown at runtime. * * @param keySerde key serde used to send key-value pairs, * if not specified the default key serde defined in the configuration will be used http://git-wip-us.apache.org/repos/asf/kafka/blob/84c995d8/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java index 8010b3a..5aafc02 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java @@ -17,6 +17,7 @@ package org.apache.kafka.streams.kstream.internals; +import org.apache.kafka.streams.errors.StreamsException; import org.apache.kafka.streams.processor.AbstractProcessor; import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorContext; @@ -54,6 +55,10 @@ public class KTableSource<K, V> implements ProcessorSupplier<K, V> { private class KTableSourceProcessor extends AbstractProcessor<K, V> { @Override public void process(K key, V value) { + // the keys should never be null + if (key == null) + throw new StreamsException("Record key for the source KTable from topic " + topic + " should not be null."); + context().forward(key, new Change<>(value, null)); } } @@ -71,11 +76,14 @@ public class KTableSource<K, V> implements ProcessorSupplier<K, V> { @Override public void process(K key, V value) { + // the keys should never be null + if (key == null) + throw new StreamsException("Record key for the source KTable from topic " + topic + " should not be null."); + V oldValue = sendOldValues ? store.get(key) : null; store.put(key, value); context().forward(key, new Change<>(value, oldValue)); } } - }
