[KAFKA-10417] suppress() with cogroup() throws ClassCastException

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 2.6.0
• Fix Version/s: 2.8.0, 2.7.1
• Component/s: streams
• Labels: kafka-streams
• Flags: Important
• Assignee: Leah Thomas
• Reporter: Wardha Perinkada Kattu
• Created: 18/Aug/20 15:05
• Updated: 16/Dec/20 14:00
• Resolved: 16/Dec/20 14:00

Description

Streams operation - `cogroup()`, `aggregate()` followed by `suppress()` throws `ClassCastException`

Works fine without the `suppress()`

Code block tested -

val stream1 = requestStreams.merge(successStreams).merge(errorStreams)
                .groupByKey(Grouped.with(Serdes.String(), serdesConfig.notificationSerde()))

        val streams2 = confirmationStreams
                .groupByKey(Grouped.with(Serdes.String(), serdesConfig.confirmationsSerde()))

        val cogrouped = stream1.cogroup(notificationAggregator).cogroup(streams2, confirmationsAggregator)
                .windowedBy(TimeWindows.of(Duration.ofMinutes(notificationStreamsConfig.joinWindowMinutes.toLong())).grace(Duration.ofMinutes(notificationStreamsConfig.graceDurationMinutes.toLong())))
                .aggregate({ null }, Materialized.`as`<String, NotificationMetric, WindowStore<Bytes, ByteArray>>("time-windowed-aggregated-stream-store")
                        .withValueSerde(serdesConfig.notificationMetricSerde()))
                 .suppress(Suppressed.untilWindowCloses(unbounded()))
                .toStream()

Exception thrown is:

Caused by: java.lang.ClassCastException: class org.apache.kafka.streams.kstream.internals.PassThrough cannot be cast to class org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier (org.apache.kafka.streams.kstream.internals.PassThrough and org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier are in unnamed module of loader 'app')

https://stackoverflow.com/questions/63459685/kgroupedstream-with-cogroup-aggregate-suppress

Issue Links

• links to: GitHub Pull Request #9727