Trying to cogroup, aggregate & suppress streams which result in -
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')
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()
However, if I remove the .suppress it works fine.