You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/01/30 23:15:04 UTC
kafka git commit: KAFKA-4613: Treat null-key records the same way for
joins and aggreations
Repository: kafka
Updated Branches:
refs/heads/trunk 413444c0e -> b559846f2
KAFKA-4613: Treat null-key records the same way for joins and aggreations
Author: Jeyhun Karimov <je...@gmail.com>
Reviewers: Matthias J. Sax, Eno Thereska, Damian Guy, Guozhang Wang
Closes #2420 from jeyhunkarimov/KAFKA-4613
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b559846f
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b559846f
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b559846f
Branch: refs/heads/trunk
Commit: b559846f2f8151e41a6259e8a570b9422932c836
Parents: 413444c
Author: Jeyhun Karimov <je...@gmail.com>
Authored: Mon Jan 30 15:14:49 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Mon Jan 30 15:14:49 2017 -0800
----------------------------------------------------------------------
.../apache/kafka/streams/kstream/KTable.java | 21 ++++++++------------
.../kstream/internals/KTableKTableJoin.java | 11 ++++------
.../kstream/internals/KTableKTableLeftJoin.java | 11 ++++------
.../internals/KTableKTableOuterJoin.java | 11 ++++------
.../internals/KTableKTableRightJoin.java | 11 ++++------
.../streams/kstream/internals/KTableSource.java | 5 ++---
.../kstream/internals/KTableKTableJoinTest.java | 8 ++++++++
.../internals/KTableKTableLeftJoinTest.java | 4 ++++
.../internals/KTableKTableOuterJoinTest.java | 4 ++++
9 files changed, 42 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
index 4b70397..77f74a8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
@@ -530,6 +530,8 @@ public interface KTable<K, V> {
* If the key or value type is changed, it is recommended to use {@link #groupBy(KeyValueMapper, Serde, Serde)}
* instead.
*
+ * If the new record key is {@code null} the record will not be included in the resulting {@link KGroupedTable}.
+ *
* @param selector a {@link KeyValueMapper} that computes a new grouping key and value to be aggregated
* @param <KR> the key type of the result {@link KGroupedTable}
* @param <VR> the value type of the result {@link KGroupedTable}
@@ -556,6 +558,8 @@ public interface KTable<K, V> {
* records to and rereading all update records from it, such that the resulting {@link KGroupedTable} is partitioned
* on the new key.
*
+ * If the new record key is {@code null} the record will not be included in the resulting {@link KGroupedTable}.
+ *
* @param selector a {@link KeyValueMapper} that computes a new grouping key and value to be aggregated
* @param keySerde key serdes for materializing this stream,
* if not specified the default serdes defined in the configs will be used
@@ -584,12 +588,9 @@ public interface KTable<K, V> {
* {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
* The key of the result record is the same as for both joining input records.
* <p>
- * Note that {@link KeyValue records} with {@code null} values (so-called tombstone records) have delete semantics.
+ * Note that {@link KeyValue records} with {@code null} keys or values (so-called tombstone records) have delete semantics.
* Thus, for input tombstones the provided value-joiner is not called but a tombstone record is forwarded
* directly to delete a record in the result {@code KTable} if required (i.e., if there is anything to be deleted).
- * <p>
- * A {@code KTable} input record key cannot be {@code null}.
- * <p>
* Example:
* <table border='1'>
* <tr>
@@ -675,13 +676,10 @@ public interface KTable<K, V> {
* null} to compute a value (with arbitrary type) for the result record.
* The key of the result record is the same as for both joining input records.
* <p>
- * Note that {@link KeyValue records} with {@code null} values (so-called tombstone records) have delete semantics.
+ * Note that {@link KeyValue records} with {@code null} keys or values (so-called tombstone records) have delete semantics.
* For example, for left input tombstones the provided value-joiner is not called but a tombstone record is
* forwarded directly to delete a record in the result {@code KTable} if required (i.e., if there is anything to be
* deleted).
- * <p>
- * A {@code KTable} input record key cannot be {@code null}.
- * <p>
* Example:
* <table border='1'>
* <tr>
@@ -768,12 +766,9 @@ public interface KTable<K, V> {
* corresponding other value to compute a value (with arbitrary type) for the result record.
* The key of the result record is the same as for both joining input records.
* <p>
- * Note that {@link KeyValue records} with {@code null} values (so-called tombstone records) have delete semantics.
+ * Note that {@link KeyValue records} with {@code null} keys or values (so-called tombstone records) have delete semantics.
* Thus, for input tombstones the provided value-joiner is not called but a tombstone record is forwarded directly
- * to delete a record in the result {@code KTable} if required (i.e., if there is anything to bedeleted).
- * <p>
- * A {@code KTable} input record key cannot be {@code null}.
- * <p>
+ * to delete a record in the result {@code KTable} if required (i.e., if there is anything to be deleted).
* Example:
* <table border='1'>
* <tr>
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoin.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoin.java
index 21a1176..f8e3f15 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoin.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoin.java
@@ -17,7 +17,6 @@
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.processor.AbstractProcessor;
@@ -76,14 +75,12 @@ class KTableKTableJoin<K, R, V1, V2> extends KTableKTableAbstractJoin<K, R, V1,
valueGetter.init(context);
}
- /**
- * @throws StreamsException if key is null
- */
@Override
public void process(final K key, final Change<V1> change) {
- // the keys should never be null
- if (key == null)
- throw new StreamsException("Record key for KTable join operator should not be null.");
+ // we do join iff keys are equal, thus, if key is null we cannot join and just ignore the record
+ if (key == null) {
+ return;
+ }
R newValue = null;
R oldValue = null;
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
index cfdaf61..ea625a7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
@@ -17,7 +17,6 @@
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.Processor;
@@ -66,14 +65,12 @@ class KTableKTableLeftJoin<K, R, V1, V2> extends KTableKTableAbstractJoin<K, R,
valueGetter.init(context);
}
- /**
- * @throws StreamsException if key is null
- */
@Override
public void process(final K key, final Change<V1> change) {
- // the keys should never be null
- if (key == null)
- throw new StreamsException("Record key for KTable left-join operator should not be null.");
+ // we do join iff keys are equal, thus, if key is null we cannot join and just ignore the record
+ if (key == null) {
+ return;
+ }
R newValue = null;
R oldValue = null;
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
index 5b32415..0809d65 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
@@ -17,7 +17,6 @@
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.Processor;
@@ -65,14 +64,12 @@ class KTableKTableOuterJoin<K, R, V1, V2> extends KTableKTableAbstractJoin<K, R,
valueGetter.init(context);
}
- /**
- * @throws StreamsException if key is null
- */
@Override
public void process(final K key, final Change<V1> change) {
- // the keys should never be null
- if (key == null)
- throw new StreamsException("Record key for KTable outer-join operator should not be null.");
+ // we do join iff keys are equal, thus, if key is null we cannot join and just ignore the record
+ if (key == null) {
+ return;
+ }
R newValue = null;
R oldValue = null;
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
index 17fa5cb..938bd43 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
@@ -17,7 +17,6 @@
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.Processor;
@@ -66,14 +65,12 @@ class KTableKTableRightJoin<K, R, V1, V2> extends KTableKTableAbstractJoin<K, R,
valueGetter.init(context);
}
- /**
- * @throws StreamsException if key is null
- */
@Override
public void process(final K key, final Change<V1> change) {
- // the keys should never be null
- if (key == null)
- throw new StreamsException("Record key for KTable right-join operator should not be null.");
+ // we do join iff keys are equal, thus, if key is null we cannot join and just ignore the record
+ if (key == null) {
+ return;
+ }
final R newValue;
R oldValue = null;
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/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 d391151..93d9de7 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,7 +17,6 @@
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;
@@ -58,9 +57,9 @@ public class KTableSource<K, V> implements ProcessorSupplier<K, V> {
@Override
public void process(K key, V value) {
- // the keys should never be null
+ // if the key is null, then ignore the record
if (key == null)
- throw new StreamsException("Record key for the source KTable from store name " + storeName + " should not be null.");
+ return;
V oldValue = store.get(key);
store.put(key, value);
tupleForwarder.maybeForward(key, value, oldValue);
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
index a938605..83cf965 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
@@ -103,6 +103,8 @@ public class KTableKTableJoinTest {
for (int i = 0; i < 2; i++) {
driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
}
+ // pass tuple with null key, it will be discarded in join process
+ driver.process(topic1, null, "SomeVal");
driver.flushState();
processor.checkAndClearProcessResult();
@@ -112,6 +114,8 @@ public class KTableKTableJoinTest {
for (int i = 0; i < 2; i++) {
driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
}
+ // pass tuple with null key, it will be discarded in join process
+ driver.process(topic2, null, "AnotherVal");
driver.flushState();
processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
@@ -165,6 +169,10 @@ public class KTableKTableJoinTest {
processor.checkAndClearProcessResult("2:XX2+YY2", "3:XX3+YY3");
checkJoinedValues(getter, kv(2, "XX2+YY2"), kv(3, "XX3+YY3"));
+
+ driver.process(topic1, null, "XX" + 1);
+ checkJoinedValues(getter, kv(2, "XX2+YY2"), kv(3, "XX3+YY3"));
+
}
@Test
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
index 9790044..9483c6b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
@@ -104,6 +104,8 @@ public class KTableKTableLeftJoinTest {
for (int i = 0; i < 2; i++) {
driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
}
+ // pass tuple with null key, it will be discarded in join process
+ driver.process(topic1, null, "SomeVal");
driver.flushState();
processor.checkAndClearProcessResult("0:X0+null", "1:X1+null");
@@ -114,6 +116,8 @@ public class KTableKTableLeftJoinTest {
for (int i = 0; i < 2; i++) {
driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
}
+ // pass tuple with null key, it will be discarded in join process
+ driver.process(topic2, null, "AnotherVal");
driver.flushState();
processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
http://git-wip-us.apache.org/repos/asf/kafka/blob/b559846f/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
index 02e4e60..65e5595 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
@@ -102,6 +102,8 @@ public class KTableKTableOuterJoinTest {
for (int i = 0; i < 2; i++) {
driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
}
+ // pass tuple with null key, it will be discarded in join process
+ driver.process(topic1, null, "SomeVal");
driver.flushState();
processor.checkAndClearProcessResult("0:X0+null", "1:X1+null");
checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, null), kv(3, null));
@@ -111,6 +113,8 @@ public class KTableKTableOuterJoinTest {
for (int i = 0; i < 2; i++) {
driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
}
+ // pass tuple with null key, it will be discarded in join process
+ driver.process(topic2, null, "AnotherVal");
driver.flushState();
processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));