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 all 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: 3 additions & 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,9 @@ 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)

@miguno miguno Jul 4, 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 to double-check because the JIRA ticket is a bit difficult to follow: Is this single condition sufficient to cover the following discussion/flow in the JIRA ticket?

  1. If "send old value" is enabled, then there are a couple of cases we can consider:

a. If old value is <key: null> and new value is <key: not-null>, and the filter predicate return false for the new value, then in this case it is safe to optimize and not returning anything to the downstream operator, since in this case we know there is no value for the key previously anyways; otherwise we send the original pair.

b. If old value is <key: not-null> and new value is <key: null>, indicating to delete this key, and the filter predicate return false for the old value, then in this case it is safe to optimize and not returning anything to the downstream operator, since we know that the old value has already been filtered in a previous message; otherwise we send the original pair.

c. If both old and new values are not null, and:

  1. predicate return true on both, send the original pair;
  2. predicate return false on both, we can optimize and do not send anything;
  3. predicate return true on old and false on new, send the key: {old -> null};
  4. predicate return false on old and true on new, send the key: {null -> new};

@miguno miguno Jul 4, 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.

With /cc @guozhangwang

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

Yes it covers all cases above, which can be summarized as if both new and old values are null.

return; // unnecessary to forward here.

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

Expand Down
39 changes: 37 additions & 2 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,8 @@
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.test.MockKeyValueMapper;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.Before;
Expand Down Expand Up @@ -254,7 +256,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 @@ -271,7 +273,40 @@ public boolean test(String key, Integer value) {
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)");
}

@Test
public void testSkipNullOnMaterialization() throws IOException {
// Do not explicitly set enableSendingOldValues. Let a further downstream stateful operator trigger it instead.
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.equalsIgnoreCase("accept");
}
}).groupBy(MockKeyValueMapper.<String, String>NoOpKeyValueMapper())
.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();
}
}
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