[
https://issues.apache.org/jira/browse/KAFKA-12564?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jessica Johann updated KAFKA-12564:
-----------------------------------
Description:
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}}
was:
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}}
> 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
> 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.
> 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)
