[ https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16586738#comment-16586738 ]
Ted Yu commented on KAFKA-7316: ------------------------------- I was thinking about removing the implicit wrapKTable . Then the following compilation errors pop up (only a snippet, there are more): {code} /Users/tyu/kafka/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala:52: type mismatch; found : org.apache.kafka.streams.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],VR] required: org.apache.kafka.streams.scala.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],VR] inner.aggregate((() => initializer).asInitializer, aggregator.asAggregator, merger.asMerger, materialized) ^ /Users/tyu/kafka/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala:64: type mismatch; found : org.apache.kafka.streams.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],Long] required: org.apache.kafka.streams.scala.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],Long] inner.count(materialized.asInstanceOf[Materialized[K, java.lang.Long, ByteArraySessionStore]]) {code} If modifying the individual places is acceptable, I can send a PR. > Use of filter method in KTable.scala may result in StackOverflowError > --------------------------------------------------------------------- > > Key: KAFKA-7316 > URL: https://issues.apache.org/jira/browse/KAFKA-7316 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 2.0.0 > Reporter: Ted Yu > Priority: Major > Labels: scala > > In this thread: > http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+ > Druhin reported seeing StackOverflowError when using filter method from > KTable.scala > This can be reproduced with the following change: > {code} > diff --git > a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala > b/streams/streams-scala/src/test/scala > index 3d1bab5..e0a06f2 100644 > --- > a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala > +++ > b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala > @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes > extends StreamToTableJ > val userClicksStream: KStream[String, Long] = > builder.stream(userClicksTopic) > val userRegionsTable: KTable[String, String] = > builder.table(userRegionsTopic) > + userRegionsTable.filter { case (_, count) => true } > // Compute the total per region by summing the individual click counts > per region. > val clicksPerRegion: KTable[String, Long] = > {code} -- This message was sent by Atlassian JIRA (v7.6.3#76005)