Skip to content

KAFKA-3902 #1556

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
phderome wants to merge 22 commits into from
Closed

KAFKA-3902 #1556

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
3827d91
remove verify left join from KStreamRepartitionJoinTest as it is caus…
dguy Jun 24, 2016
e07388b
KAFKA-3902
phderome Jun 26, 2016
3563044
KAFKA-3902
phderome Jun 26, 2016
721e00d
KAFKA-3902
phderome Jun 26, 2016
0a93d16
Make the JoinWindow in KStreamRepartitionJoinTest much larger (1 minu…
dguy Jun 27, 2016
1919fa5
Merge branch 'kafka-3896' of https://github.com/dguy/kafka into DEROM…
phderome Jun 27, 2016
aa036cb
Merge branch 'trunk' of https://github.com/apache/kafka into DEROME-3902
phderome Jun 27, 2016
a607e24
avoid unconditional materialization as per Guozhang's explanation and…
phderome Jun 28, 2016
e6beae8
Fix test cases accordingly so only the oldValue null, null are suppre…
phderome Jun 28, 2016
a8f9ef7
clears state after a check as it used to be in the first place, thus …
phderome Jun 28, 2016
7c277cf
revert the double null filter addition in KTableFilter and add a mean…
phderome Jun 29, 2016
73479d3
revert the double null filter addition in KTableFilter and add a mean…
phderome Jun 29, 2016
b039e4e
keep Guozhang's 2nd fix
phderome Jun 29, 2016
356b0a6
suppress nulls more aggressively than originally discussed, seems mor…
phderome Jun 30, 2016
66aada0
Merge branch 'trunk' of https://github.com/apache/kafka into DEROME-3902
phderome Jun 30, 2016
b2f5c06
suppress nulls more aggressively than originally discussed, seems mor…
phderome Jun 30, 2016
0fe41c8
revert back to Guozhang's suggestions.
phderome Jun 30, 2016
5b19a71
revert back to Guozhang's suggestions.
phderome Jul 1, 2016
c1e4ddf
Merge branch 'trunk' of https://github.com/apache/kafka into DEROME-3902
phderome Jul 1, 2016
117c661
removed unnecessary/incorrect comments.
phderome Jul 1, 2016
525d1b4
simplified unit test testSkipNullOnMaterialization as per feedback.
phderome Jul 1, 2016
1dccdba
simplified unit test testSkipNullOnMaterialization as per feedback.
phderome Jul 1, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ class KTableFilter<K, V> implements KTableProcessorSupplier<K, V, V> {
private final Predicate<K, V> predicate;
private final boolean filterNot;

private boolean sendOldValues = false;
private boolean sendOldValues = true;

@guozhangwang guozhangwang Jun 27, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My concern is that, the overhead of requesting the source KTable to be materialized (i.e. creating a state store, and sending the {old -> new} pair instead of the new value only) may be over-whelming compared with its potential benefits of reducing the downstream traffic.

@phderome phderome Jun 28, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am looking for your guidance here. On the one hand, I think your point has merit and have difficulty assessing the tradeoff given limited knowledge of app usage and app internals; on the other hand, I'd be very curious to hear what comparable projects do in similar situation (I read Samza's demonstration of materialized tables and user clicks example, so perhaps you have familiarity with Samza's or others' design decisions if in sync with K-Streams, you might have worked on that project as well). Specifically, I wonder if the rationale you had presented for case 2 in the JIRA statement (sendOldValues=false) really requires sending keys with nulls on false evaluation of filter (is that how Samza would do it too?). If case 2's rationale can be challenged, then this ticket can have a meaningful outcome, otherwise it seems that this ticket can not lead to a behaviour change.

I'd be happy to reset this to false if you find that more advisable. But then, I don't see any hook at higher level of abstraction to enable sending old values so that for instance RegionView example from Confluent would work without a final filter to exclude nulls on Long deserializer.

@guozhangwang guozhangwang Jun 28, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Samza does not have a high-level DSL interface, so it dos not have this issue.

This issue is related to the more general question that when should we materialize a KTable object with a state store, and currently the rule is that stateless operators like filters should not necessarily materialize it. That being said, if a further downstream operator is stateful, it will cause backward propagation of setting enableSendOldValues to true up till the source KTable, as I mentioned in the email thread.


public KTableFilter(KTableImpl<K, ?, V> parent, Predicate<K, V> predicate, boolean filterNot) {
this.parent = parent;
Expand Down Expand Up @@ -77,6 +77,7 @@ public void process(K key, Change<V> change) {
V newValue = computeValue(key, change.newValue);

@guozhangwang guozhangwang Jun 27, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we add a check on the original change value that newValue and oldValue cannot be both null?

@phderome phderome Jun 28, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean something semantically equivalent to
if (change.oldValue == null && change.newValue == null) return;
or rather
if (sendOldValues && change.oldValue == null && change.newValue == null) return;

I'd like to understand whether you're addressing case 2 alone or case 2 and 3. Once I am clear on what you mean, I can make change and test.

@guozhangwang guozhangwang Jun 28, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually never mind, I was originally thinking that after this patch we should if (change.oldValue == null && change.newValue == null) throw StreamsException, since it should not happen; but I just realized a mapValues operator can still make both values to be null and pass it to the downstream filter.

V oldValue = sendOldValues ? computeValue(key, change.oldValue) : null;

if (sendOldValues && oldValue == null && newValue == null) return; // unnecessary to forward here.
context().forward(key, new Change<>(newValue, oldValue));
}

Expand Down
18 changes: 9 additions & 9 deletions streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,8 +91,8 @@ public boolean test(String key, Integer value) {
driver.process(topic1, "A", null);
driver.process(topic1, "B", null);

proc2.checkAndClearProcessResult("A:null", "B:2", "C:null", "D:4", "A:null", "B:null");
proc3.checkAndClearProcessResult("A:1", "B:null", "C:3", "D:null", "A:null", "B:null");
proc2.checkAndClearProcessResult("B:2", "D:4");
proc3.checkAndClearProcessResult("A:1", "C:3");
}

@Test
Expand Down Expand Up @@ -203,7 +203,7 @@ public boolean test(String key, Integer value) {
driver.process(topic1, "C", 1);

proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)", "C:(null<-null)");
proc2.checkEmpty();

driver.process(topic1, "A", 2);
driver.process(topic1, "B", 2);
Expand All @@ -214,13 +214,13 @@ public boolean test(String key, Integer value) {
driver.process(topic1, "A", 3);

proc1.checkAndClearProcessResult("A:(3<-null)");
proc2.checkAndClearProcessResult("A:(null<-null)");
proc2.checkEmpty();

driver.process(topic1, "A", null);
driver.process(topic1, "B", null);

proc1.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)");
proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)");
proc2.checkEmpty();
}

@Test
Expand Down Expand Up @@ -254,24 +254,24 @@ public boolean test(String key, Integer value) {
driver.process(topic1, "C", 1);

proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)", "C:(null<-null)");
proc2.checkEmpty(); // we got nothing since all inputs are odd or filtered out

driver.process(topic1, "A", 2);
driver.process(topic1, "B", 2);

proc1.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)");
proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)"); // we are informed of 2 making it in for both A and B

driver.process(topic1, "A", 3);

proc1.checkAndClearProcessResult("A:(3<-2)");
proc2.checkAndClearProcessResult("A:(null<-2)");
proc2.checkAndClearProcessResult("A:(null<-2)"); // no change for B but A is deleted

driver.process(topic1, "A", null);
driver.process(topic1, "B", null);

proc1.checkAndClearProcessResult("A:(null<-3)", "B:(null<-2)");
proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-2)");
proc2.checkAndClearProcessResult("B:(null<-2)"); // B is deleted from source Table1
}

}
2 changes: 1 addition & 1 deletion streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ public boolean test(String key, Integer value) {

assertEquals(Utils.mkList("A:01", "B:02", "C:03", "D:04"), proc1.processed);
assertEquals(Utils.mkList("A:1", "B:2", "C:3", "D:4"), proc2.processed);
assertEquals(Utils.mkList("A:null", "B:2", "C:null", "D:4"), proc3.processed);
assertEquals(Utils.mkList("B:2", "D:4"), proc3.processed);
assertEquals(Utils.mkList("A:01", "B:02", "C:03", "D:04"), proc4.processed);
}

Expand Down
4 changes: 4 additions & 0 deletions streams/src/test/java/org/apache/kafka/test/MockProcessorSupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,10 @@ public void checkAndClearProcessResult(String... expected) {
processed.clear();
}

public void checkEmpty() {
assertEquals("the number of outputs:", 0, processed.size());
}

public void checkAndClearPunctuateResult(long... expected) {
assertEquals("the number of outputs:", expected.length, punctuated.size());

Expand Down