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

Matthias J. Sax commented on KAFKA-15417:
-----------------------------------------

Thanks. I did look into this and found this PR: 
https://github.com/apache/kafka/commit/03411ca28b21d554d754b3f67a7857afe64f5bef

It introduces `joinSpuriousLookBackTimeMs` but it seems it's set before/after 
for the left/right join incorrectly – I think it must be flipped... The added 
unit tests seems to make incorrect assumptions when a window should get closed, 
but it was not caught during review (it's easy to flip before/after in your 
mind...)

Hence, the fix itself is simple, but updating the unit test is a little bit 
more work. Would you be interested to work on a PR to fix it?

> JoinWindow does not  seem to work properly with a KStream - KStream - 
> LeftJoin()
> --------------------------------------------------------------------------------
>
>                 Key: KAFKA-15417
>                 URL: https://issues.apache.org/jira/browse/KAFKA-15417
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 3.4.0
>            Reporter: Victor van den Hoven
>            Priority: Major
>         Attachments: SimpleStreamTopology.java, SimpleStreamTopologyTest.java
>
>
> In Kafka-streams 3.4.0 :
> According to the javadoc of the Joinwindow:
> _There are three different window configuration supported:_
>  * _before = after = time-difference_
>  * _before = 0 and after = time-difference_
>  * _*before = time-difference and after = 0*_
>  
> However if I use a joinWindow with *before = time-difference and after = 0* 
> on a kstream-kstream-leftjoin the *after=0* part does not seem to work.
> When using _stream1.leftjoin(stream2, joinWindow)_ with 
> {_}joinWindow{_}.{_}after=0 and joinWindow.before=30s{_}, any new message on 
> stream 1 that can not be joined with any messages on stream2 should be joined 
> with a null-record after the _joinWindow.after_ has ended and a new message 
> has arrived on stream1.
> It does not.
> Only if the new message arrives after the value of _joinWindow.before_ the 
> previous message will be joined with a null-record.
>  
> Attached you can find two files with a TopologyTestDriver Unit test to 
> reproduce.
> topology:   stream1.leftjoin( stream2, joiner, joinwindow)
> joinWindow has before=5000ms and after=0
> message1(key1) ->  stream1
> after 4000ms message2(key2) -> stream1  ->  NO null-record join was made, but 
> the after period was expired.
> after 4900ms message2(key2) -> stream1  ->  NO null-record join was made, but 
> the after period was expired.
> after 5000ms message2(key2) -> stream1  ->  A null-record join was made,  
> before period was expired.
> after 6000ms message2(key2) -> stream1  ->  A null-record join was made,  
> before period was expired.
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Reply via email to