[jira] [Commented] (KAFKA-6599) KTable KTable join semantics violated when caching enabled

["Alexey Osipov (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-6599?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17298101#comment-17298101 ] 

Alexey Osipov commented on KAFKA-6599:
--------------------------------------

Stepped into this bug recently using kafka version 2.7.0.

The point I want to emphasize is that this bug is not only connected with oldValue propagation. The thing I'm concerned most is, when caching enabled, KTable-KTable join produces incorrect result in case both A and B are set to null in a short time frame.

Here is the expected timeline for join result updates given the A and B values:
||Timestamp||A||B||A join B||
|0|null|null|null|
|1|A| |(no change - still null)|
|2| |B|AB|
|3|null| |null|

 Here is what I see in reality when caching enabled:
||Timestamp||A||B||A join B||
|0|null|null|null|
|1|A| |(no change - still null)|
|2| |B|AB|
|3|null| |{color:#FF0000}(no change - still AB){color}|
|4| |null|{color:#FF0000}(no change - still AB){color}|
|5| | |{color:#FF0000}(no change - still AB){color}|
|6| | |{color:#FF0000}...{color}|

 The problem is that "null" updates in timestamps 3 and 4 are got cached. When "commit.interval.ms" expires, those nulls start propagation downstream. In "org.apache.kafka.streams.kstream.internals.KTableKTableInnerJoin.KTableKTableJoinProcessor#process", when A's "null" update is processed, it tries to get value of B for the same key by calling "valueGetter.get(key)". That call returns null, because B's value for the same key is already "null" (*cached*). Since B is null, downstream propagation doesn't happen:

 
{noformat}
if (valueRight == null) {
    return;
}{noformat}
Same thing happens when B's "null" update is processed. It does lookup for A's value, it returns null from cache, and propagation doesn't happen.

 

*Without any further activity in A and B on that key, "A join B" permanently remains "AB" (while it should be null).*

CC [~damianguy] [~mjsax] Not sure if this is you guys who I should ping on this, I just found your names in similar tickets (KAFKA-4609, KAFKA-9916). Please redirect this information to whoever you think is appropriate. Thanks!

> KTable KTable join semantics violated when caching enabled
> ----------------------------------------------------------
>
>                 Key: KAFKA-6599
>                 URL: https://issues.apache.org/jira/browse/KAFKA-6599
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>            Reporter: Jan Filipiak
>            Priority: Critical
>              Labels: bug
>
> Say a tuple A,B got emmited after joining and the delete for A goes into the cache. After that the B record would be deleted aswell. B's join processor would look up A and see `null` while computing for old and new value (at this point we can execute joiner with A beeing null and still emit something, but its not gonna represent the actual oldValue) Then As cache flushes it doesn't see B so its also not gonna put a proper oldValue. The output can then not be used for say any aggregate as a delete would not reliably find its old aggregate where it needs to be removed from filter will also break as it stopps null,null changes from propagating. So for me it looks pretty clearly that Caching with Join breaks KTable semantics. be it my new join or the currently existing once.
>  
> this if branch here
> [https://github.com/apache/kafka/blob/1.0/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java#L155]
> is not usefull. I think its there because when one would delegate the true case to the underlying. One would get proper semantics for streams, but the weiredest cache I've seen.
>  



--
This message was sent by Atlassian Jira
(v8.3.4#803005)