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 16 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
1 change: 1 addition & 0 deletions 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 @@ -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 (oldValue == null && newValue == null) return; // unnecessary to forward here.

@guozhangwang guozhangwang Jun 30, 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.

I'm not sure if we can optimize the case when sendOldValues is false as well. Following your example:

KTable T1 = builder.table("source-topic");
KTable T2 = table.filter(value > 2);
T2.to("sink-topic");

And suppose the "source-topic" is piping the messages to T1 as: {a: 3}, {b: 5}, {a: 1}...

When {a: 3} is passed from T1 to T2, the filter will pass and hence it is forwarded to downstream operators already; so now when later {a: 1} is passed from T1 to T2, meaning "modifying the value with key {a} from 3 to 1", the filter will not pass any more, and hence in this case we need to forward a {a: null} record downstreams in order to indicate the previously forwarded {a: 3} has now been deleted in T2, right? Otherwise the sink topic will have the following messages:

{a: 3}, {b: 5}, ...

whereas the right sequence should be

{a: 3}, {b: 5}, {a: null}, ...

@phderome phderome Jul 1, 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.

Clear explanation which I validated with a temporary test that proved indeed it was a wrong idea. The deletion under not copying old value failed to occur with my earlier commit.

So, it's back to original solution with return on 2nd line and a unit test to demonstrate usage of sendOldValues indirectly via materialization caused by aggregation.

context().forward(key, new Change<>(newValue, oldValue));
}

Expand Down
66 changes: 57 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 @@ -24,6 +24,10 @@
import org.apache.kafka.streams.kstream.Predicate;
import org.apache.kafka.test.KStreamTestDriver;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockReducer;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.Before;
Expand Down Expand Up @@ -91,8 +95,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 +207,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.checkEmptyAndClearProcessResult();

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

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

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.checkEmptyAndClearProcessResult();
}

@Test
Expand Down Expand Up @@ -254,24 +258,68 @@ 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.checkEmptyAndClearProcessResult(); // 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
}

@Test
public void testSkipNullOnMaterialization() throws IOException {
// do not explicitly set enableSendingOldValues, which is considered a cheat. Let a further downstream stateful operator trigger it instead.
// This test may well need review/deletion with future materialization changes, should materialization occur more frequently for instance.
KStreamBuilder builder = new KStreamBuilder();

String topic1 = "topic1";

KTableImpl<String, String, String> table1 =
(KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1);
KTableImpl<String, String, String> table2 = (KTableImpl<String, String, String>) table1.filter(
new Predicate<String, String>() {
@Override
public boolean test(String key, String value) {
return value.compareToIgnoreCase("accept") == 0;

@guozhangwang guozhangwang Jul 1, 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 use equalsIgnoreCase directly.

@phderome phderome Jul 1, 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.

will do.

}
}).mapValues(
new ValueMapper<String, String>() {

@guozhangwang guozhangwang Jul 1, 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.

Is this mapValues really needed?

@phderome phderome Jul 1, 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.

Taking it out, but keeping the groupBy that follows.

@Override
public String apply(String value) {
return value;
}
}).groupBy(new KeyValueMapper<String, String, KeyValue<String, String>>() {
@Override
public KeyValue<String, String> apply(String first, String second) {
return (second == null || first.compareTo(second) <= 0) ? new KeyValue<>(first, first) : new KeyValue(second, second);

@guozhangwang guozhangwang Jul 1, 2016 •
edited
Loading

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.

Just use equalsTo? Also I thought the aggregation is only needed for enforce sending old values? In this case can we just use a org.apache.kafka.test.MockAggregator?

@phderome phderome Jul 1, 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 can make groupBy as following, not testing meaningfully first, second (we don't care, do we?)
return new KeyValue<>(first, first);

@guozhangwang guozhangwang Jul 1, 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.

I said it incorrectly before, and I actually meant that you can use org.apache.kafka.test.NoOpKeyValueMapper in this case as we do not really care about what aggregate key / value to use. Your proposal is also fine, while I was just trying to reduce duplicated code :)

}
}).reduce(MockReducer.STRING_ADDER, MockReducer.STRING_REMOVER, "mock-result");

MockProcessorSupplier<String, String> proc1 = new MockProcessorSupplier<>();
MockProcessorSupplier<String, String> proc2 = new MockProcessorSupplier<>();

builder.addProcessor("proc1", proc1, table1.name);
builder.addProcessor("proc2", proc2, table2.name);

driver = new KStreamTestDriver(builder, stateDir, stringSerde, stringSerde);

driver.process(topic1, "A", "reject");
driver.process(topic1, "B", "reject");
driver.process(topic1, "C", "reject");

proc1.checkAndClearProcessResult("A:(reject<-null)", "B:(reject<-null)", "C:(reject<-null)");
proc2.checkEmptyAndClearProcessResult(); // we got nothing since no input matches (though enableSendingOldValues is not set by test on table2 explicitly)
}
}
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
6 changes: 6 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,12 @@ public void checkAndClearProcessResult(String... expected) {
processed.clear();
}

public void checkEmptyAndClearProcessResult() {

assertEquals("the number of outputs:", 0, processed.size());
processed.clear();
}

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

Expand Down