KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-77) #1777

mjsax · 2016-08-23T22:37:38Z

fixed leftJoin -> outerJoin test bug
simplified to only use values
fixed inner KTable-KTable join
fixed left KTable-KTable join
fixed outer KTable-KTable join
fixed inner, left, and outer left KStream-KStream joins
added inner KStream-KTable join
fixed left KStream-KTable join

guozhangwang · 2016-08-26T05:22:24Z

I tried to re-run the unit tests locally and seems the following three test cases fail frequently:

org.apache.kafka.streams.integration.JoinIntegrationTest > testInnerKTableKTable FAILED
    java.lang.AssertionError: Condition not met within timeout 300000. Did not receive 1 number of records
        at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:256)
        at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinValuesRecordsReceived(IntegrationTestUtils.java:245)
        at org.apache.kafka.streams.integration.JoinIntegrationTest.checkResult(JoinIntegrationTest.java:169)
        at org.apache.kafka.streams.integration.JoinIntegrationTest.runTest(JoinIntegrationTest.java:196)
        at org.apache.kafka.streams.integration.JoinIntegrationTest.testInnerKTableKTable(JoinIntegrationTest.java:366)

org.apache.kafka.streams.integration.JoinIntegrationTest > testLeftKTableKTable FAILED
    java.lang.AssertionError: Condition not met within timeout 300000. Did not receive 1 number of records
        at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:256)
        at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinValuesRecordsReceived(IntegrationTestUtils.java:245)
        at org.apache.kafka.streams.integration.JoinIntegrationTest.checkResult(JoinIntegrationTest.java:169)
        at org.apache.kafka.streams.integration.JoinIntegrationTest.runTest(JoinIntegrationTest.java:196)
        at org.apache.kafka.streams.integration.JoinIntegrationTest.testLeftKTableKTable(JoinIntegrationTest.java:393)

org.apache.kafka.streams.kstream.internals.KTableKTableLeftJoinTest > testNotSendingOldValue FAILED
    java.lang.AssertionError
        at org.junit.Assert.fail(Assert.java:86)
        at org.junit.Assert.assertTrue(Assert.java:41)
        at org.junit.Assert.assertTrue(Assert.java:52)
        at org.apache.kafka.streams.kstream.internals.KTableKTableLeftJoinTest.testNotSendingOldValue(KTableKTableLeftJoinTest.java:181)

mjsax · 2016-08-26T12:55:23Z

@guozhangwang Just realized, that I forget to commit something which fails left-join tests...

About inner-join: that is exactly the timing issues we discussed already. The integration test is super slow... not sure why it takes to long for the result consumer to receive output records...
If you observe system-out you will see that it take multiple seconds -- and sometime even longer...

mjsax · 2016-08-27T16:48:23Z

The last run passed (ie, no timeout). However, the integration test takes quite long (to process 15 records...).

https://builds.apache.org/job/kafka-trunk-git-pr-jdk7/5304/testReport/org.apache.kafka.streams.integration/JoinIntegrationTest/

guozhangwang · 2016-08-29T16:16:39Z

@mjsax When you got time, could you do some profiling on the inner join integration test to see what is the performance bottleneck? This latency is surprising and may indicate some bugs AFAIK.

mjsax · 2016-09-29T02:38:00Z

Please review @hjafarpour @guozhangwang @enothereska @dguy @miguno

dguy

One general comment before I look further. There appear to be quite a few changes that are related to format changes - did you mean to do this? The problem with these sort of changes is that we probably all have different settings in our IDEs, so any formatting rules applied can end up causing unnecessary changes, conflicts etc.

mjsax · 2016-09-29T16:02:27Z

@dguy I just use default Intellij formatting with some adjustments to meet code style checks of the build -- I thought that is the default we use? No? I don't care about the formatting, but I strongly recommend that we agree on a unique formatting and use the same auto-formatting rules...

Completely agree, that formatting changes in PR should be avoided.

dguy

I'm struggling to find what has actually changed as most of the changes are formatting and adding final to parameters etc. While i agree with using final everywhere possible, do we really want to be doing it as part of this PR?

dguy · 2016-09-30T15:57:43Z

clients/src/test/java/org/apache/kafka/test/TestUtils.java

        }

-        if (!testCondition.conditionMet()) {
+        if (!testConditionMet) {


Is this meant to be in this PR? I think i saw that you've done it in another?

I extracted this into the other PR -- need to update this one...

mjsax · 2016-09-30T16:07:34Z

@dguy I can revert all formatting changes -- however, I am wondering if anyone would do a "reformat only" PR? I guess if we want to get unique auto-formatted code, we need to do this gradually over all PRs. WDYT?

dguy · 2016-09-30T16:26:19Z

Yes, i'm not sure anyone would do a reformat only PR. However, i think the best approach is as you suggested, do it gradually, i.e., just changes in the area you are currently working in. Otherwise it becomes to much noise to process, i.e., hence the struggle to review the real changes

mjsax · 2016-09-30T18:16:14Z

@dguy updated (reverted reformatting changes and removed test improvement)

dguy

Some minor comments, but otherwise LGTM

dguy · 2016-10-03T09:51:38Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

+        }
+    }
+
+    private <V1, R> KStream<K, R> doStreamTableJoin(final KTable<K, V1> other,


Could we add another param to this method, boolean leftJoin and then get rid of doStreamTableLeftJoin?

dguy · 2016-10-03T09:53:02Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

+
+    private <V1, R> KStream<K, R> doStreamTableJoin(final KTable<K, V1> other,
+                                                    final ValueJoiner<V, V1, R> joiner) {
+        final Set<String> allSourceNodes = ensureJoinableWith((AbstractStream<K>) other);


Should we start adding @SuppressWarnings("unchecked") to the method for these unchecked casts?

dguy · 2016-10-03T09:56:38Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java

+
+        @Override
+        public void process(final K key, final V1 value) {
+            // if the key is null, we do not need proceed joining


Probably should update or remove the comment.

dguy · 2016-10-03T10:03:34Z

streams/src/test/java/org/apache/kafka/streams/integration/JoinIntegrationTest.java

+        CLUSTER.deleteTopic(INPUT_TOPIC_2);
+        CLUSTER.deleteTopic(OUTPUT_TOPIC);
+
+        TestUtils.waitForCondition(topicsGotDeleted, 300, "Topics not deleted after 120 seconds.");


The wait time is 300 milliseconds, yet the message says 120 seconds...

dguy · 2016-10-03T10:07:38Z

streams/src/test/java/org/apache/kafka/streams/integration/JoinIntegrationTest.java

+    private KTable<Long, String> leftTable;
+    private KTable<Long, String> rightTable;
+
+    private final long anyUniqueKey = 0L;


Do we need this field? It is only used in the constructor of Input

guozhangwang · 2016-10-05T00:20:18Z

One meta comment about the re-formatting is that, mixing multiple "changes" in one PR will make the reviews harder, as there will be more diffs to look at that may not be directly related to the core change.

guozhangwang

Overall LGTM. One meta suggestion: let's leave some comments in each of the join processor explaining the join semantics, regarding null values etc.

guozhangwang · 2016-10-05T23:23:48Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java


-        return join(other, joiner, windows, null, null, null, false);
+        return doJoin(other, joiner, windows, null, null, null, new KStreamImplJoin(false, false));


nit: call join(other, joiner, windows, null, null, null) instead?

guozhangwang · 2016-10-05T23:24:12Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java


-        return join(other, joiner, windows, null, null, null, true);
+        return doJoin(other, joiner, windows, null, null, null, new KStreamImplJoin(true, true));


nit: call outerJoin(other, joiner, windows, null, null, null) instead?

guozhangwang · 2016-10-05T23:29:45Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java

-        public void process(K key, V1 value) {
-            if (key == null)
+        public void process(final K key, final V1 value) {
+            if (key == null || value == null) {


maybe add a comment explaining why we are excluding <key: null> for stream stream join, pros and cons compared with treating each <key: null> as a different message?

guozhangwang · 2016-10-05T23:33:16Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

@@ -666,12 +668,12 @@ public boolean test(final K1 key, final V1 value) {
                                                   Serde<V2> otherValueSerde) {
            String thisWindowStreamName = topology.newName(WINDOWED_NAME);
            String otherWindowStreamName = topology.newName(WINDOWED_NAME);
-            String joinThisName = outer ? topology.newName(OUTERTHIS_NAME) : topology.newName(JOINTHIS_NAME);
-            String joinOtherName = outer ? topology.newName(OUTEROTHER_NAME) : topology.newName(JOINOTHER_NAME);
+            String joinThisName = leftOuter ? topology.newName(OUTERTHIS_NAME) : topology.newName(JOINTHIS_NAME);


should we check on rightOuter in line 671 and leftOuter in line 672?

Is not "this join" the left hand side of the join and "other join" the right hand side of the join? Can you clarify?

joinThisName is used in the store of thisWindow which is to be queried by the other stream, so my personal understanding is that:

for inner-join (rightOuter = false, leftOuter = false): both window-store has JOINTHIS_NAME-store.

for outer-join (rightOuter = true, leftOuter = true): both window-store has OUTERTHIS_NAME-store.

for left-join (rightOuter = false, leftOuter = true): the left window-store THIS_NAME-store and the right window-store OUTERTHIS_NAME-store, since we will join with null if the right window-store returns null (hence "outer"), but not vice-versa.

guozhangwang · 2016-10-05T23:34:29Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java

            // if the key is null, we do not need proceed joining
            // the record with the table
-            if (key != null) {
+            if (key != null && value != null) {


ditto above: maybe add a comment explaining why we are excluding <key: null> for stream stream join.

guozhangwang · 2016-10-05T23:36:11Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java

                context().forward(key, joiner.apply(value, valueGetter.get(key)));
            }
        }
    }

+    private class KStreamKTableJoinProcessor extends AbstractProcessor<K, V1> {


Could we merge KStreamKTableJoinProcessor with KStreamKTableLeftJoinProcessor into a single processor?

guozhangwang · 2016-10-05T23:40:57Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java

@@ -312,6 +312,8 @@ public V apply(Change<V> change) {
        Objects.requireNonNull(joiner, "joiner can't be null");

        Set<String> allSourceNodes = ensureJoinableWith((AbstractStream<K>) other);
+        enableSendingOldValues();


Why we want to enforce sending old data for table-table join now?

We need old value to omit duplicate null tombstone result records.

I agree that we need to require all the parent processors of the join operator to send old values, originally I thought we were requiring the join processor to also send old values but checking again I think you are doing the right thing: only require parent KTable streams to send old values, so this is actually correct.

guozhangwang · 2016-10-05T23:44:30Z

...ams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java

-                    "[D@0]:null",
-                    "[A@0]:null"
-            );
+            proc3.checkAndClearProcessResult();


Why we need to change the test for windowed aggregation here?

We do test inner KTable-KTable join and null values are ignored. (I cannot recall exactly: according to KIP, this should have been empty an result before this change -- however, it seem correct this way to me.)

sounds good.

guozhangwang · 2016-10-05T23:46:09Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java


-        // push all items to the other stream. this should produce four items.
+        // push all items to the other stream. this should produce two items.


Still four items, right?

mjsax · 2016-10-08T06:11:40Z

Updated and rebase.

DO NOT MERGE (need to rebase again after #1992 got merged)

guozhangwang · 2016-10-18T05:04:52Z

@mjsax #1992 is merged, could you rebase again?

mjsax · 2016-10-18T05:41:37Z

@guozhangwang Rebased.

guozhangwang · 2016-10-19T04:42:49Z

Let's merge #2039 first, left a comment on the PR for unit test.

- fixed leftJoin -> outerJoin test bug - simplified to only use values - fixed inner KTable-KTable join - fixed left KTable-KTable join - fixed outer KTable-KTable join - fixed inner, left, and outer left KStream-KStream joins - added inner KStream-KTable join - fixed left KStream-KTable join

mjsax · 2016-10-20T17:48:50Z

@guozhangwang rebases as #2039 got merged.

guozhangwang · 2016-10-20T20:06:49Z

Merged to trunk.

mjsax force-pushed the kafka-4001-joins branch 3 times, most recently from cb2f1e1 to cca15e3 Compare August 24, 2016 00:32

mjsax changed the title ~~KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-76) [WiP]~~ KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-77) [WiP] Aug 27, 2016

mjsax force-pushed the kafka-4001-joins branch from b4de1d4 to 8382cd0 Compare August 28, 2016 15:32

mjsax force-pushed the kafka-4001-joins branch from 8382cd0 to 1782b5b Compare August 31, 2016 15:50

mjsax changed the title ~~KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-77) [WiP]~~ KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-77) Sep 5, 2016

mjsax force-pushed the kafka-4001-joins branch 2 times, most recently from 953074a to 6a6f214 Compare September 28, 2016 20:19

dguy reviewed Sep 29, 2016

View reviewed changes

mjsax mentioned this pull request Sep 29, 2016

KAFKA-4058: improved test condition evaluation #1938

Closed

dguy reviewed Sep 30, 2016

View reviewed changes

mjsax force-pushed the kafka-4001-joins branch from 6a6f214 to 36a1712 Compare September 30, 2016 18:13

dguy reviewed Oct 3, 2016

View reviewed changes

guozhangwang reviewed Oct 5, 2016

View reviewed changes

mjsax force-pushed the kafka-4001-joins branch from 36a1712 to b3459b9 Compare October 8, 2016 06:09

mjsax force-pushed the kafka-4001-joins branch from b3459b9 to 60efd27 Compare October 18, 2016 05:41

mjsax added 6 commits October 19, 2016 23:32

increased timeout

e5f4318

reverted reformatting changes

5e238bf

addressed Damian's comments

8125ec7

addressed Guozhang's comments

58e5fd8

rebased

e417b1c

mjsax force-pushed the kafka-4001-joins branch from 60efd27 to e417b1c Compare October 20, 2016 17:48

asfgit closed this in 62c0972 Oct 20, 2016

mjsax deleted the kafka-4001-joins branch October 21, 2016 00:03


		return join(other, joiner, windows, null, null, null, false);
		return doJoin(other, joiner, windows, null, null, null, new KStreamImplJoin(false, false));


		// push all items to the other stream. this should produce four items.
		// push all items to the other stream. this should produce two items.

KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-77) #1777

KAFKA-4001: Improve Kafka Streams Join Semantics (KIP-77) #1777

Conversation

mjsax commented Aug 23, 2016 • edited

guozhangwang commented Aug 26, 2016 • edited

mjsax commented Aug 26, 2016

mjsax commented Aug 27, 2016

guozhangwang commented Aug 29, 2016

mjsax commented Sep 29, 2016

dguy left a comment

Choose a reason for hiding this comment

mjsax commented Sep 29, 2016

dguy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Sep 30, 2016

dguy commented Sep 30, 2016

mjsax commented Sep 30, 2016 • edited

dguy 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

Choose a reason for hiding this comment

guozhangwang commented Oct 5, 2016

guozhangwang 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

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

mjsax commented Oct 8, 2016

guozhangwang commented Oct 18, 2016

mjsax commented Oct 18, 2016

guozhangwang commented Oct 19, 2016

mjsax commented Oct 20, 2016

guozhangwang commented Oct 20, 2016

mjsax commented Aug 23, 2016 •

edited

guozhangwang commented Aug 26, 2016 •

edited

mjsax commented Sep 30, 2016 •

edited