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

[Jessica Johann (Jira)]

Jessica Johann created KAFKA-12564:
--------------------------------------

             Summary: 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: Jessica Johann


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. 
Feed the Stream "issue_stream" with:

{{(1->"A")}}
{{(1->"B")}}
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. 
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. 
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)