[
https://issues.apache.org/jira/browse/KAFKA-7778?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16775681#comment-16775681
]
ASF GitHub Bot commented on KAFKA-7778:
---------------------------------------
casey-green commented on pull request #6314: KAFKA-7778: Add KTable.suppress to
Scala API
URL: https://github.com/apache/kafka/pull/6314
### Detailed description
Adds `KTable.suppress` to the Scala API.
NOTE: `KTable.count.suppress` **does not work**! If I replace
`.aggregate(0L)((_, _, agg) => agg + 1)` with `count()` in my unit tests, I get
this exception:
```
Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to [B
at
org.apache.kafka.common.serialization.ByteArraySerializer.serialize(ByteArraySerializer.java:19)
at
org.apache.kafka.streams.kstream.internals.FullChangeSerde$1.serialize(FullChangeSerde.java:73)
at
org.apache.kafka.streams.kstream.internals.FullChangeSerde$1.serialize(FullChangeSerde.java:60)
at
org.apache.kafka.streams.kstream.internals.suppress.KTableSuppressProcessor.buffer(KTableSuppressProcessor.java:96)
at
org.apache.kafka.streams.kstream.internals.suppress.KTableSuppressProcessor.process(KTableSuppressProcessor.java:87)
at
org.apache.kafka.streams.kstream.internals.suppress.KTableSuppressProcessor.process(KTableSuppressProcessor.java:40)
at
org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
```
This appears to be because the value serde is not getting passed down to the
`KTable` in count:
https://github.com/casey-green/kafka/blob/trunk/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedTable.scala#L50.
Subsequent `KTable`s then fallback to the default serde (byte array), which is
incorrect. I tried fiddling with this code a bit to get it to work, to no avail.
### Testing strategy
I added unit tests requested in this PR:
https://github.com/apache/kafka/pull/6092
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Add KTable.suppress to Scala API
> --------------------------------
>
> Key: KAFKA-7778
> URL: https://issues.apache.org/jira/browse/KAFKA-7778
> Project: Kafka
> Issue Type: Task
> Components: streams
> Affects Versions: 2.1.0
> Reporter: Jacek Laskowski
> Assignee: John Roesler
> Priority: Major
> Labels: newbie
>
> {{KTable.suppress}} is not available in Scala API.
> [https://cwiki.apache.org/confluence/display/KAFKA/KIP-328%3A+Ability+to+suppress+updates+for+KTables]
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
