[jira] [Commented] (KAFKA-12564) KTable#filter-method called twice after aggregation

["Jess J. (Jira)" <ji...@apache.org>]

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

Jess J. commented on KAFKA-12564:
---------------------------------

[~mjsax] Thanks for your explanation and time. (y)

> KTable#filter-method called twice after aggregation
> ---------------------------------------------------
>
>                 Key: KAFKA-12564
>                 URL: https://issues.apache.org/jira/browse/KAFKA-12564
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 2.7.0
>            Reporter: Jess J.
>            Priority: Major
>
> Libraries from build.sbt:
> {{"org.apache.kafka" % "kafka_2.13" % "2.7.0",}}
> {{"org.apache.kafka" % "kafka-streams" % "2.7.0",}}
> {{"org.apache.kafka" % "kafka-clients" % "2.7.0",}}
> {{"org.apache.kafka" % "kafka-streams-scala_2.13" % "2.7.0",}}
> h4.  
> h4. Feed the Stream "issue_stream" with:
> {{(1->"A")}}
>  {{(1->"B")}}
> h4.  
> h4. Topology:
> {{// #1}}
>  {{val issueStream:KStream[Int,String] = builder.stream[Int,String]("issue_stream")}}
>  
> {{// #2}}
>  {{val aggTable:KTable[Int,String] =}}
>  {{issueStream}}
>  {{.groupBy((k,v)=>k)}}
>  {{.aggregate[String]("EMPTY")((k,v,agg)=>s"$agg+$v")}}
>  
> {{// #3}}
>  {{aggTable}}
>  {{.toStream}}
>  {{.print(Printed.toSysOut)}}
>  
> {{// #4}}
>  {{aggTable.filter((k,v)=> {}}
>  {{  println(s"filter($k, $v) at ${System.nanoTime()}")}}
>  {{  true}}
>  {{})}}
>  {{.toStream}}
>  {{.print(Printed.toSysOut)}}
> h4.  
> h4. First Tuple: (1->"A")
> #3 Output as expected, the aggregated tuple ("EMPTY"+"+A")
> {{[KTABLE-TOSTREAM-0000000044]: 1, EMPTY+A}}
>  
> #4 The filter-method is called twice.
>  First call with the expected tuple.
> {{filter(1, EMPTY+A) at 211379567071847}}
> The second call with the empty initialized aggregate.
> {{filter(1, EMPTY) at 211379567120375}}
> Output contains the correct tuple
> {{[KTABLE-TOSTREAM-0000000047]: 1, EMPTY+A}}
> h4.  
> h4. Second Tuple: (1->"B")
> #3 Output as expected the aggregated tuple ("EMPTY"+"+A"+"+B")
> {{[KTABLE-TOSTREAM-0000000044]: 1, EMPTY+A+B}}
> #4 Again a second unexpected call to filter(...) with the previous tuple before aggregation
>  First call:
> {{filter(1, EMPTY+A+B) at 211379567498482}}
> Second call:
> {{filter(1, EMPTY+A) at 211379567524475}}
> But the output contains only one tuple as expected
> {{[KTABLE-TOSTREAM-0000000047]: 1, EMPTY+A+B}}



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