KIP-557: Add Emit On Change Support by ConcurrencyPractitioner · Pull Request #8254 · apache/kafka

ConcurrencyPractitioner · 2020-03-08T18:29:07Z

This is the initial draft PR for adding emit on change. For reference of the design document, please see here.

https://cwiki.apache.org/confluence/display/KAFKA/KIP-557%3A+Add+emit+on+change+support+for+Kafka+Streams

Committer Checklist (excluded from commit message)

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

ConcurrencyPractitioner · 2020-03-08T18:31:07Z

@vvcephei @cadonna @mjsax Would you recommend this approach for implementing binary equality? I just want some feedback on this, since it seems there are multiple ways to resolve this issue,

vvcephei · 2020-03-09T19:58:37Z

test this please

vvcephei · 2020-03-09T19:59:27Z

Thanks for this PR @ConcurrencyPractitioner ! I'll take a look at this asap. Just wanted to acknowledge that it's on my plate.

vvcephei · 2020-03-12T16:29:23Z

Just FYI, it looks like the tests couldn't build.

ConcurrencyPractitioner · 2020-03-13T02:48:43Z

@vvcephei Acknowledged. My intention wasn't really to get a working model of the code. I was intending this to be a draft of what the code might look like. I just want approval on this approach first before I dive in.

ConcurrencyPractitioner · 2020-03-14T00:25:51Z

@vvcephei It looks like casting the store to MeteredTimestampedKeyValueStore doesn't work, as in some cases, the TimestampedKeyValueStore is in fact TimestampedKeyValueStoreReadWriteDecorator. So we will have to abandon this approach. In this case, we will probably need to consider passing in null serdes instead.

…tioner/kafka into EMIT-ON-CHANGE

ConcurrencyPractitioner · 2020-03-29T21:30:24Z

@vvcephei @mjsax @guozhangwang Pinging for initial review.

vvcephei

Hey @ConcurrencyPractitioner , thanks for the PR! I think this looks really close. I'm recommending a few tweaks.

I think you can convert this WIP PR to an actual PR by addressing my comments and adding some additional test cases.

Thanks again!
-John

…ls/KTableSourceTest.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

…ls/KTableSource.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

…/MeteredTimestampedKeyValueStore.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

vvcephei

Thanks for the update. I just noticed one small thing I missed last time.

@mjsax or @cadonna , do you have a chance to take a look at this? I believe this is what we agreed on for the KIP, but want to double-check with you.

vvcephei · 2020-04-02T20:41:02Z

+            final byte[] serializedValue = wrapped().get(keyBytes(key));
+            return new RawAndDeserializedValue<V>(serializedValue,
+                maybeMeasureLatency(() -> outerValue(serializedValue), time, getSensor));


Ah, missed this the last time though. We should also perform the wrapped().get inside the lambda, so that the time to perform the get is included in the latency.

Cool, I will make the change.

vvcephei · 2020-04-03T16:00:26Z

test this please

vvcephei · 2020-04-03T17:51:18Z

test this please

vvcephei · 2020-04-03T17:51:26Z

test this please

vvcephei · 2020-04-29T18:18:32Z

test this please

vvcephei · 2020-04-29T18:18:39Z

test this please

vvcephei · 2020-04-29T18:18:47Z

test this please

vvcephei · 2020-04-29T18:18:55Z

retest this please

ConcurrencyPractitioner · 2020-04-29T23:57:04Z

@vvcephei Looks like I forgot to push. You can retrigger the tests after this one.

ConcurrencyPractitioner · 2020-04-30T02:56:50Z

@vvcephei Alright, all tests passed (looks like there were some flakies) and all comments addressed.

ConcurrencyPractitioner · 2020-05-06T02:44:44Z

@vvcephei Just pinging. I looked into the flakies, and it turns out they were different test failures for each build. So its the equivalent of an all green.

vvcephei · 2020-05-11T20:41:27Z

Retest this please

vvcephei · 2020-05-11T20:42:07Z

Sorry, for the delay @ConcurrencyPractitioner . I have been insanely busy. I really appreciate your patience.

I'm giving it a final pass now.

vvcephei

Hey, @ConcurrencyPractitioner ,

Again, I'm really sorry there have been so many delays. I have just a few very quick remarks. If you don't have time to address them (or don't care to wait on me anymore ;) ), I can also do it while merging.

Thanks again!

vvcephei · 2020-05-11T20:54:02Z

+        assertFalse(metered.putIfDifferentValues(key,
+                                                 newValueAndTimestamp,
+                                                 encodedOldValue));


Ah, sorry to say, one more thing slipped by me before. We should verify(inner) at the end of both of these tests. It should actually fail for shouldNotPutIfSameValuesAndGreaterTimestamp because we should not call inner.put in this case. To fix that, we would just delete L202, where we set up the mock for inner.put. Then, the mock would be initialized to expect no calls, and the verification would fail if we did call it.

vvcephei · 2020-05-11T20:54:30Z

+        final byte[] encodedOldValue = stringSerde.serializer().serialize("TOPIC", valueAndTimestamp);
+
+        final ValueAndTimestamp<String> outOfOrderValueAndTimestamp = ValueAndTimestamp.make("value", 95L);
+        assertTrue(metered.putIfDifferentValues(key, outOfOrderValueAndTimestamp, encodedOldValue));


As I mentioned in the earlier comment, we're missing a verify(inner) call at the end of this test.

vvcephei · 2020-05-11T21:00:12Z

+     *              if the timestamp of right is less than left (indicating out of order record)
+     *         false otherwise
+     */
+    public static boolean maskTimestampAndCompareValues(final byte[] left, final byte[] right) {


Ah, upon reading that last test, I realized that I previously overlooked when you added the timestamp comparison to this method. We should change the method name for maintainability. It no longer just "masks the timestamp and compares the values". Can we instead call it "compareValuesAndCheckForIncreasingTimestamp" or something? I can almost guarantee that one or more people will be badly misled by the current method name.

ConcurrencyPractitioner · 2020-05-11T23:54:14Z

@vvcephei Alright, cool. Got those last comments addressed.

vvcephei · 2020-05-12T18:07:07Z

Unrelated test failures:

kafka.api.SslConsumerTest.testCoordinatorFailover

org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true]

kafka.api.ConsumerBounceTest.testClose

org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication

vvcephei · 2020-05-12T18:13:33Z

Hey @ConcurrencyPractitioner , I just pushed a very minor test fix. I'm verifying it locally before merging.

vvcephei · 2020-05-12T18:20:28Z

Thanks for the awesome contribution, @ConcurrencyPractitioner !

* 'trunk' of github.com:apache/kafka: MINOR: add option to rebuild source for system tests (apache#6656) KAFKA-9850 Move KStream#repartition operator validation during Topolo… (apache#8550) MINOR: Add a duplicate() method to Message classes (apache#8556) KAFKA-9966: add internal assignment listener to stabilize eos-beta upgrade test (apache#8648) MINOR: Replace null with an actual value for timestamp field in InsertField SMT unit tests (apache#8649) MINOR: Fix ProcessorContext JavaDocs and stream-time computation (apache#8603) MINOR: improve tests for TopologyTestDriver (apache#8631) KAFKA-9821: consolidate Streams rebalance triggering mechanisms (apache#8596) KAFKA-9669; Loosen validation of inner offsets for older message formats (apache#8647) KAFKA-8770: KIP-557: Drop idempotent KTable source updates (apache#8254) MINOR: Remove allow concurrent test (apache#8641)

rodesai · 2020-05-14T07:57:11Z

+                    if (ValueAndTimestampSerializer.compareValuesAndCheckForIncreasingTimestamp(oldSerializedValue, newSerializedValue)) {
+                        return false;
+                    } else {
+                        wrapped().put(keyBytes(key), newSerializedValue);


@ConcurrencyPractitioner @vvcephei I'm trying to understand this to debug some broken tests in ksql. Couple questions:

When the timestamp of the newer value is lower (ignoring the value), why do we want to put the new value into the store? Surely the store should have the value with the newer timestamp? Otherwise we could wind up with a corrupt store.

Don't we still want to put the value in the store (even if we don't forward it on to the next context) if the values are the same but the timestamp is newer? Otherwise if we get an out-of-order update with a different value, but a timestamp in between the rows with the same value, we'd incorrectly put that value into the store, e.g. the following updates:

TS: 1, K: X, V: A
TS: 3, K: X, V: A
TS: 2, K: X, V: B

would result in the table containing K: X, V: B, which is wrong.

Why when the timestamp of the newer value is lower, do we want to put the new value into the store? Surely the store should have the value with the newer timestamp? Otherwise we could wind up with a corrupt store.

This behavior was there also before this PR. If a out-of-order record is encountered, a log message was written, but the record was nevertheless put into the state store (cf.

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

Line 122 in 7624e62

store.put(key, ValueAndTimestamp.make(value, context().timestamp()));

). The only thing that changed is that if the serialized value of the new record is equal to the serialized value of the old value and the timestamp of the new record is equal or newer, we drop the record because it is a idempotent update.
Could you elaborate on why a store should get corrupted because of this?

would result in the table containing K: X, V: B, which is wrong.

As said above, this behavior should not have been changed.

Don't we still want to put the value in the store (even if we don't forward it on to the next context) if the values are the same but the timestamp is newer?

If we just put the value in the store but did not forward it, then the store would actually be corrupted, because the local state would not be consistent with downstream anymore.

Not putting a record with the same value but a newer timestamp in the store and not forwarding it was the main point of this KIP.

Drops idempotent updates from KTable source operators. Specifically, drop updates in which the value is unchanged, and the timestamp is the same or larger. Implements: KIP-557 Reviewers: Bruno Cadonna <bruno@confluent.io>, John Roesler <vvcephei@apache.org>

The latest commit #8254 on this test deleted all topics after each test, but the topic was actually shared among tests before. And after that we are relying on the less-reliable auto-topic generation to get the topic which makes the test flaky. I'm now using different topics for different tests, also setting the app.id for tests differently. Reviewers: Boyang Chen <boyang@confluent.io>, A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>

Drops idempotent updates from KTable source operators. Specifically, drop updates in which the value is unchanged, and the timestamp is the same or larger. Implements: KIP-557 Reviewers: Bruno Cadonna <bruno@confluent.io>, John Roesler <vvcephei@apache.org>

[KIP-557] Add emit on change support

80de5bb

vvcephei added the streams label Mar 9, 2020

ConcurrencyPractitioner added 8 commits March 13, 2020 20:37

Adding some class modifications

1aaaa01

Adding class

ec46f57

Merge branch 'trunk' into EMIT-ON-CHANGE

2c19bf7

Adding test

e3ffa93

Merge branch 'EMIT-ON-CHANGE' of https://github.com/ConcurrencyPracti…

a14725a

…tioner/kafka into EMIT-ON-CHANGE

Adding sensor

6d3c7ab

Add working sensor

81c19bd

Bumping processor level

209d376

vvcephei reviewed Apr 1, 2020

View reviewed changes

ConcurrencyPractitioner and others added 6 commits April 2, 2020 08:59

Update streams/src/test/java/org/apache/kafka/streams/kstream/interna…

8fef99c

…ls/KTableSourceTest.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

Update streams/src/main/java/org/apache/kafka/streams/kstream/interna…

1d7c4ad

…ls/KTableSource.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

Update streams/src/main/java/org/apache/kafka/streams/state/internals…

d062bfa

…/MeteredTimestampedKeyValueStore.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

Fixing compilation errors and other issues

d0737f5

Removing unneeded class

afdde96

Removing all vestiges of new class

35c16b1

ConcurrencyPractitioner changed the title ~~KIP-557: Add Emit On Change Support [WIP]~~ KIP-557: Add Emit On Change Support Apr 2, 2020

vvcephei reviewed Apr 2, 2020

View reviewed changes

Adding more inclusive lambda

9e7649b

ConcurrencyPractitioner added 3 commits April 29, 2020 08:34

Deleting massive amount of print statements

527ba28

Final removal

bae2860

Resolving last comment

bf5532f

vvcephei reviewed May 11, 2020

View reviewed changes

Addressing some last comments

d9aa12d

vvcephei reviewed May 12, 2020

View reviewed changes

Comment thread .../test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java Outdated

minor test fix

2a3d93d

vvcephei merged commit f54cece into apache:trunk May 12, 2020

rodesai reviewed May 14, 2020

View reviewed changes

guozhangwang mentioned this pull request Jun 10, 2020

KAFKA-9991: Fix flaky unit tests #8843

Merged

3 tasks

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

ConcurrencyPractitioner mentioned this pull request Jun 16, 2020

KIP-557: Add emit on change support to Kafka Streams #8881

Closed

Uh oh!

Conversation

ConcurrencyPractitioner commented Mar 8, 2020

Committer Checklist (excluded from commit message)

Uh oh!

ConcurrencyPractitioner commented Mar 8, 2020

Uh oh!

vvcephei commented Mar 9, 2020

Uh oh!

vvcephei commented Mar 9, 2020

Uh oh!

vvcephei commented Mar 12, 2020

Uh oh!

ConcurrencyPractitioner commented Mar 13, 2020

Uh oh!

ConcurrencyPractitioner commented Mar 14, 2020

Uh oh!

ConcurrencyPractitioner commented Mar 29, 2020

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei Apr 2, 2020

Choose a reason for hiding this comment

Uh oh!

ConcurrencyPractitioner Apr 3, 2020

Choose a reason for hiding this comment

Uh oh!

vvcephei commented Apr 3, 2020

Uh oh!

vvcephei commented Apr 3, 2020

Uh oh!

vvcephei commented Apr 3, 2020

Uh oh!

vvcephei commented Apr 29, 2020

Uh oh!

vvcephei commented Apr 29, 2020

Uh oh!

vvcephei commented Apr 29, 2020

Uh oh!

vvcephei commented Apr 29, 2020

Uh oh!

ConcurrencyPractitioner commented Apr 29, 2020

Uh oh!

ConcurrencyPractitioner commented Apr 30, 2020

Uh oh!

ConcurrencyPractitioner commented May 6, 2020

Uh oh!

vvcephei commented May 11, 2020

Uh oh!

vvcephei commented May 11, 2020

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei May 11, 2020

Choose a reason for hiding this comment

Uh oh!

vvcephei May 11, 2020

Choose a reason for hiding this comment

Uh oh!

vvcephei May 11, 2020

Choose a reason for hiding this comment

Uh oh!

ConcurrencyPractitioner commented May 11, 2020

Uh oh!

vvcephei commented May 12, 2020

Uh oh!

Uh oh!

vvcephei commented May 12, 2020

Uh oh!

vvcephei commented May 12, 2020

Uh oh!

rodesai May 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

rodesai May 14, 2020 •

edited

Loading

cadonna May 14, 2020 •

edited

Loading

cadonna May 14, 2020 •

edited

Loading