KAFKA-6455: Extend CacheFlushListener to forward timestamp by mjsax · Pull Request #6147 · apache/kafka

mjsax · 2019-01-15T19:45:57Z

This PR is of part of the overall KIP-258 story. It does not resolve KAFKA-6455, but prepares it. It's only internal changes so we can merge right away.

mjsax · 2019-01-15T19:46:17Z

Call for review @guozhangwang @bbejeck @vvcephei

mjsax

Marked the relevant changes.

All others are to keep the code working :)

mjsax · 2019-01-15T19:46:56Z

+     * @param timestamp   timestamp of new value
     */
-    void apply(final K key, final V newValue, final V oldValue);
+    void apply(final K key, final V newValue, final V oldValue, final long timestamp);


This is the actual change.

mjsax · 2019-01-15T19:47:28Z

-package org.apache.kafka.streams.kstream.internals;
-
-import org.apache.kafka.streams.state.internals.ThreadCache;
+package org.apache.kafka.streams.state.internals;


Minor fix: The interface should be in state package, not kstream.

Good catch!

mjsax · 2019-01-15T19:47:53Z

+                    serdes.keyFrom(entry.key().get()),
+                    serdes.valueFrom(entry.newValue()),
+                    oldValue,
+                    entry.entry().context().timestamp());


part of the fix: we now also forward the timestamp on eviction

mjsax · 2019-01-15T19:47:56Z

+                        key,
+                        newValue,
+                        oldValue,
+                        entry.entry().context().timestamp());


part of the fix: we now also forward the timestamp on eviction

mjsax · 2019-01-15T19:48:01Z

+                    windowedKey,
+                    serdes.valueFrom(entry.newValue()),
+                    oldValue,
+                    entry.entry().context().timestamp());


part of the fix: we now also forward the timestamp on eviction

mjsax · 2019-01-16T00:17:43Z

Retest this please.

bbejeck

Thanks for PR @mjsax, left a couple of comments.

bbejeck · 2019-01-16T19:09:05Z

+                }
+                forward(child, key, value);
+            } else {
+                if (children.size() == 1) {


I think this if/else block here can be removed and just go with

for (final ProcessorNode child : children) { forward(child, key, value); }

I copied this from ProcessorContextImpl#forward() -- we added it there are "optimization" to avoid creating the iterator -- might be pre-mature optimization to be fair.

I don't feel strong about it, however, I would prefer to keep both classed "in sync". Thus, if we remove it here, I think we should also remove it from there?

Throughts?

Yeah if it exists elsewhere let's just leave it as is for now.

Let's remove on both side: I think in J8 it is not really a big difference.

Ok. Will simplify this.

bbejeck · 2019-01-16T19:12:35Z

-            for (final ProcessorNode child : (List<ProcessorNode<K, V>>) currentNode().children()) {
-                setCurrentNode(child);
-                child.process(key, value);
+            final List<ProcessorNode<K, V>> children = (List<ProcessorNode<K, V>>) currentNode().children();


We should add a test or tests for this section

Thinking about this once more, can a "global processor" have multiple children to begin with? Maybe we can simplify this method significantly? Thoughts?

Will add a test for GlobalProcessorContextImpl though -- there is not at all atm.

Thinking about this once more, can a "global processor" have multiple children to begin with?

Yeah, you could be correct about, I don't think it can.

+1 on assuming a single children, check-and-throw-otherwise

mjsax · 2019-01-17T02:35:26Z

Updated this.

mjsax · 2019-01-17T19:03:00Z

Retest this please.

guozhangwang

Pleas feel free to merge after comments addressed.

guozhangwang · 2019-01-17T23:16:14Z

-            for (final ProcessorNode child : (List<ProcessorNode<K, V>>) currentNode().children()) {
-                setCurrentNode(child);
-                child.process(key, value);
+            final List<ProcessorNode<K, V>> children = (List<ProcessorNode<K, V>>) currentNode().children();


+1 on assuming a single children, check-and-throw-otherwise

guozhangwang · 2019-01-17T23:17:08Z

+                }
+                forward(child, key, value);
+            } else {
+                if (children.size() == 1) {


Let's remove on both side: I think in J8 it is not really a big difference.

guozhangwang · 2019-01-17T23:19:11Z

-package org.apache.kafka.streams.kstream.internals;
-
-import org.apache.kafka.streams.state.internals.ThreadCache;
+package org.apache.kafka.streams.state.internals;


Good catch!

guozhangwang · 2019-01-17T23:24:33Z

 import org.junit.Before;
 import org.junit.Test;

+import java.util.ArrayList;


Shouldn't we add new test case similar to the ones in GlobalProcessorContextImplTest.java in this class as well?

Why? ProcessorContextImpl was not changed.

Ack, you right.

mjsax · 2019-01-18T00:58:24Z

@bbejeck @guozhangwang I was thinking about this once more, and for global tasks, we know that there are exactly two processors: the source processor and the "state maintainer" processor. The first one has exactly one child and it calls forward() from above. The second processor only call this forward overload on flush and should have zero children.

Does this make sense?

There is a related JIRA about that but whether we'd keep it as is still open questions, I think we can make this assumption still atm but just bring it up FYI.

https://issues.apache.org/jira/browse/KAFKA-7125

mjsax · 2019-01-18T00:59:41Z

The changes in this class are just a code simplification as discussed with @bbejeck and @guozhangwang -- no change in behavior.

mjsax · 2019-01-18T01:00:18Z

Updated this.

mjsax · 2019-01-18T18:47:20Z

Retest this please.

\cc For review @ableegoldman

mjsax · 2019-01-18T21:53:16Z

Test failure in core. Retest this please.

mjsax · 2019-01-19T06:02:18Z

Retest this please.

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>

* ak/trunk: MINOR: fix race condition in KafkaStreamsTest (apache#6185) KAFKA-4850: Enable bloomfilters (apache#6012) MINOR: ducker-ak: add down -f, avoid using a terminal in ducker test KAFKA-5117: Stop resolving externalized configs in Connect REST API MINOR: Cleanup handling of mixed transactional/idempotent records (apache#6172) KAFKA-7844: Use regular subproject for generator to fix *All targets (apache#6182) Fix Documentation for cleanup.policy is out of date (apache#6181) MINOR: increase timeouts for KafkaStreamsTest (apache#6178) MINOR: Rejoin split ssl principal mapping rules (apache#6099) MINOR: Handle case where connector status endpoints returns 404 (apache#6176) MINOR: Remove unused imports, exceptions, and values (apache#6117) KAFKA-3522: Add internal RecordConverter interface (apache#6150) Fix Javadoc of KafkaConsumer (apache#6155) KAFKA-6455: Extend CacheFlushListener to forward timestamp (apache#6147) MINOR: Log partition info when creating new request batch in controller (apache#6145) KAFKA-7652: Part I; Fix SessionStore's findSession(single-key) (apache#6134) MINOR: Remove the InvalidTopicException handling in InternalTopicManager (apache#6167) [KAFKA-7024] Rocksdb state directory should be created before opening the DB (apache#6138) MINOR:: Fix typos (apache#6079)

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>

KAFKA-6455: Extend CacheFlushListener to forward timestamp

7ec1b55

mjsax added the streams label Jan 15, 2019

mjsax commented Jan 15, 2019

View reviewed changes

bbejeck reviewed Jan 16, 2019

View reviewed changes

Github comments

91d4596

guozhangwang approved these changes Jan 17, 2019

View reviewed changes

mjsax commented Jan 18, 2019

View reviewed changes

Github comments

30b8447

mjsax force-pushed the minor-extend-flushlistener branch from 8e74612 to 30b8447 Compare January 18, 2019 03:06

mjsax merged commit e19eb3e into apache:trunk Jan 19, 2019

mjsax deleted the minor-extend-flushlistener branch January 19, 2019 17:40

abbccdda pushed a commit to abbccdda/kafka that referenced this pull request Jan 24, 2019

KAFKA-6455: Extend CacheFlushListener to forward timestamp (apache#6147)

83651d3

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>

mjsax mentioned this pull request May 8, 2019

KAFKA-6521: Use timestamped stores for KTables #6667

Merged

pengxiaolong pushed a commit to pengxiaolong/kafka that referenced this pull request Jun 14, 2019

KAFKA-6455: Extend CacheFlushListener to forward timestamp (apache#6147)

8620293

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>

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

Conversation

mjsax commented Jan 15, 2019

Uh oh!

mjsax commented Jan 15, 2019

Uh oh!

mjsax 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!

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!

mjsax commented Jan 16, 2019

Uh oh!

bbejeck 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!

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!

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!

mjsax commented Jan 17, 2019

Uh oh!

mjsax commented Jan 17, 2019

Uh oh!

guozhangwang 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!

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!

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!

mjsax commented Jan 18, 2019

Uh oh!

mjsax commented Jan 18, 2019

Uh oh!

mjsax commented Jan 18, 2019

Uh oh!

mjsax commented Jan 19, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects