MINOR: Improve Join integration test coverage, PART I by guozhangwang · Pull Request #4331 · apache/kafka

guozhangwang · 2017-12-15T22:26:16Z

Rename JoinIntegrationTest to StreamStreamJoinIntegrationTest, which is only for KStream-KStream joins.
Extract the AbstractJoinIntegrationTest which is going to be used for all the join integration test classes, parameterized with and without caching.
Merge KStreamRepartitionJoinTest.java into StreamStreamJoinIntegrationTest.java with augmented stream-stream join.
Add TableTableJoinIntegrationTest with detailed per-step expected results and removed KTableKTableJoinIntegrationTest.

Findings of the integration test:

Confirmed KAFKA-4309 with caching turned on.
Found bug KAFKA-6398.
Found bug KAFKA-6443.
Found a bug that in CachingKeyValueStore, we would flush before putting the record into the underlying store, when the store is going to be used in the downstream processors with flushing it would result in incorrect results, fixed the issue along with this PR.
Consider a new optimization described in KAFKA-6286.

Future works including stream-table joins will be in other PRs.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…into KMinor-join-integration-tests

…oin-integration-tests

guozhangwang · 2018-01-11T18:23:43Z

-
                final V oldValue = sendOldValues ? serdes.valueFrom(underlying.get(entry.key())) : null;
+
+                underlying.put(entry.key(), entry.newValue());


This is intentional as a bug fix. See description.

Wouldn't we want to have an else clause that still performs underlying.put in case flushListener is null?

Ack, and I think this is the root cause of the jenkins failure.

guozhangwang · 2018-01-11T18:42:03Z

@mjsax @bbejeck

bbejeck · 2018-01-11T21:55:39Z

+    @Rule
+    public final TemporaryFolder testFolder = new TemporaryFolder(TestUtils.tempDirectory());
+
+    @Parameterized.Parameters


IMHO we should consider changing to @Parameterized.Parameters(name = "caching enabled = {0}") which prints the whether caching is enabled or not vs. just the index of the parameter.

Good point!

bbejeck · 2018-01-11T21:56:47Z

+    void runTest(final List<List<String>> expectedResult, final String storeName) throws Exception {
+        assert expectedResult.size() == input.size();
+
+        System.out.println(builder.build().describe());


is this intentional?

bbejeck · 2018-01-11T21:57:16Z

+            TestUtils.waitForCondition(new TestCondition() {
+                @Override
+                public boolean conditionMet() {
+                    System.out.println("RESULT: " + finalResultReached.get());


leftover debugging?

bbejeck · 2018-01-11T22:06:42Z

left some comments.

bbejeck · 2018-01-11T22:22:42Z

@guozhangwang I think the failure could be related (or maybe a rebase is needed), I can reproduce locally.

guozhangwang · 2018-01-12T01:16:52Z

@bbejeck Thanks for reminding, the leftover debugging was my bad..

…oin-integration-tests

guozhangwang · 2018-01-12T05:36:18Z

retest this please

bbejeck

Thanks @guozhangwang LGTM

guozhangwang · 2018-01-12T17:41:13Z

Local runs saved about 40 seconds for the streams unit test suite.

dguy · 2018-01-12T18:21:50Z

            if (flushListener != null) {
-
                final V oldValue = sendOldValues ? serdes.valueFrom(underlying.get(entry.key())) : null;
+                underlying.put(entry.key(), entry.newValue());


Is this actually needed? If a downstream processor tries to get this value won't it get it from the cache? i.e., i don't think an evicted entry is removed from the cache until after flush has finished

Actually it will be removed, i.e. when listener.apply(entries); is called the entry is no longer in the cache anymore.

What I observed originally is an issue when caching is turned off: note that with that case we still go through this code path (which should be optimized away anyways in the future I think), when you call a put on the store, it will immediately trigger flush and hence be processed downstream while it is not put into the underlying and also not in the cache any more.

dguy · 2018-01-12T18:22:19Z

+
+            checkResult(OUTPUT_TOPIC, expectedFinalResult, numRecordsExpected);
+
+            if (storeName != null)


nit: java style

dguy · 2018-01-12T18:22:27Z

+                }
+            }
+
+            if (storeName != null)


dguy

Thanks @guozhangwang. Left a couple of comments, but overall LGTM. Feel free to merge once you've addressed them

dguy · 2018-01-12T18:26:20Z

+        assertThat(onlyEntry.value, is(expectedFinalResult));
+        assertThat(all.hasNext(), is(false));
+
+        all.close();


this will never be called if one of the assertions fails

Good point!

guozhangwang · 2018-01-12T23:41:22Z

Thanks for your reviews. Merged to trunk.

1. In the caching layer's flush listener call, we should always write to the underlying store, before flushing (see #4331 's point 4) for detailed explanation). When fixing 4331, it only touches on KV stores, but it turns out that we should fix for window and session store as well. 2. Also apply the optimization that was in session-store already: when the new value bytes and old value bytes are all null (this is possible e.g. if there is a put(K, V) followed by a remove(K) or put(K, null) and these two operations only hit the cache), upon flushing this mean the underlying store does not have this value at all and also no intermediate value has been sent to downstream as well. We can skip both putting a null to the underlying store as well as calling the flush listener sending `null -> null` in this case. Modifies corresponding unit tests. Reviewers: John Roesler <john@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>

1. In the caching layer's flush listener call, we should always write to the underlying store, before flushing (see apache#4331 's point 4) for detailed explanation). When fixing 4331, it only touches on KV stores, but it turns out that we should fix for window and session store as well. 2. Also apply the optimization that was in session-store already: when the new value bytes and old value bytes are all null (this is possible e.g. if there is a put(K, V) followed by a remove(K) or put(K, null) and these two operations only hit the cache), upon flushing this mean the underlying store does not have this value at all and also no intermediate value has been sent to downstream as well. We can skip both putting a null to the underlying store as well as calling the flush listener sending `null -> null` in this case. Modifies corresponding unit tests. Reviewers: John Roesler <john@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>

guozhangwang added 13 commits November 29, 2017 17:01

ingested stream stream join

c1e00d0

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

71f2625

…into KMinor-join-integration-tests

create a base abstract join class

4d95bdb

Add more tests

defd7ed

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

2a55c66

…into KMinor-join-integration-tests

consider using mock time

73f58fb

Merge branch 'trunk' of https://github.com/apache/kafka into KMinor-j…

d210de3

…oin-integration-tests

fix a bug in flushing ordering

ae2ba11

Merge branch 'trunk' of https://github.com/apache/kafka into KMinor-j…

4d5e80f

…oin-integration-tests

rebase from trunk

eb5f995

Add more tests to table table join

d495386

Merge branch 'trunk' of https://github.com/apache/kafka into KMinor-j…

a93ee09

…oin-integration-tests

minor fixes

a7bb2e8

guozhangwang commented Jan 11, 2018

View reviewed changes

guozhangwang added 4 commits January 11, 2018 10:24

remove debugging prints

33ef938

remove old table table join integration test

2c33f17

rename functions

e01a072

more renames and minor fixes

1ee24db

guozhangwang changed the title ~~[WIP] MINOR: Improve Join integration test coverage, PART I~~ MINOR: Improve Join integration test coverage, PART I Jan 11, 2018

guozhangwang added 3 commits January 11, 2018 10:53

fix style

6285217

fix style again..

b89cc99

add a FIXME comment

3728b8f

bbejeck reviewed Jan 11, 2018

View reviewed changes

github comments

f660fc5

Merge branch 'trunk' of https://github.com/apache/kafka into KMinor-j…

3d2499c

…oin-integration-tests

bbejeck approved these changes Jan 12, 2018

View reviewed changes

dguy reviewed Jan 12, 2018

View reviewed changes

dguy approved these changes Jan 12, 2018

View reviewed changes

github comments

c2dde66

guozhangwang closed this in 3e21e17 Jan 12, 2018

guozhangwang mentioned this pull request Jan 24, 2019

KAFKA-7652: Part III; Put to underlying before Flush #6191

Merged

3 tasks

guozhangwang deleted the KMinor-join-integration-tests branch April 24, 2020 23:47


		final V oldValue = sendOldValues ? serdes.valueFrom(underlying.get(entry.key())) : null;

		underlying.put(entry.key(), entry.newValue());


		checkResult(OUTPUT_TOPIC, expectedFinalResult, numRecordsExpected);

		if (storeName != null)

Conversation

guozhangwang commented Dec 15, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 11, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Jan 11, 2018

Uh oh!

bbejeck commented Jan 11, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

guozhangwang commented Jan 12, 2018

Uh oh!

guozhangwang commented Jan 12, 2018

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 12, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 12, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

guozhangwang commented Dec 15, 2017 •

edited

Loading

bbejeck commented Jan 11, 2018 •

edited

Loading