inferred type arguments [?1,?0] do not conform to method map's type parameter bounds [KR,VR]

[sjhan...@mailbox.org]

I am translating kafka's streams word count example to scala version. 

https://github.com/apache/kafka/blob/trunk/streams/examples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountDemo.java

But compiler complains inferred type arguments do not conform to map function's type parameter bounds, and type mismatch.

inferred type arguments [?1,?0] do not conform to method map's type parameter bounds [KR,VR]

[error]         ).map(

[error]           ^

type mismatch;

[error]  found   : org.apache.kafka.streams.kstream.KeyValueMapper[String,String,org.apache.kafka.streams.KeyValue[String,String]]

[error]  required: org.apache.kafka.streams.kstream.KeyValueMapper[_ >: String, _ >: String, _ <: org.apache.kafka.streams.KeyValue[_ <: KR, _ <: VR]]

[error]           new KeyValueMapper[String, String, KeyValue[String, String]] {

[error]           ^

The code snippet looks like

        val counts = source.flatMapValues(

          new ValueMapper[String, java.lang.Iterable[String]] {

            override def apply(value: String): java.lang.Iterable[String] = {

              val ary = value.toLowerCase(Locale.getDefault()).split(" ")

              Arrays.asList(ary).asInstanceOf[java.lang.Iterable[String]]

            }

          }

        ).map(

          new KeyValueMapper[String, String, KeyValue[String, String]] {

            override def apply(key: String, value: String): KeyValue[String, String] = new KeyValue(value, value)            

          }

        ).groupByKey().count("Counts");

How can I fix this compilation error?

Thanks. 

[oss.m...@gmail.com]

Arrays.asList(ary).asInstanceOf[java.lang.Iterable[String]]

This is wrong. asList takes varags I think. You have to spread ary. See, asInstanceOf can kill you. But it doesn't have to be the only problem. It might need more explicit type parameters because of call site variance.

This should help a bit:

override def apply(value: String): java.lang.Iterable[String] = {

  val ary = value.toLowerCase(Locale.getDefault()).split(" ")

  Arrays.asList(ary: _*)
}

Petr

[oss.m...@gmail.com]

Without spread it pass your Array[String] as the first arg returning Iterable[Array[String]]. Not a good idea to do asInstanceOf.

Petr

[oss.m...@gmail.com]

KR, VR are unknown. What is the type of "source"? I would guess it is KStream[Nothing, Nothing] and that you use default serdes and have plain buider.stream("my_topic"). Scala can't infer it because this way the information is on runtime. Either annotate "source" when doing builder.source[String, String]("my_topic") or use the variant which provides serdes explicitly - builder.source(Serdes.String(), Serdes.String(), "my_topic").

It could be fine then. If not find where types are inferred to Nothing. Java patterns do mess to Scala type inference.

Petr

[robert towne]

Did you ever find out how to solve this - I'm running into the same problem as well..

Thanks!

Robert

[Adriaan Moors]

Hi,

The following snippet compiles for me on Scala 2.12.1 (against kafka-streams 0.10.2.0).

Thanks IntelliJ for translating most of the Java code :-)

Full source and sbt build: https://gist.github.com/adriaanm/b0104d88ba86e918d4464ce30b829606

cheers

adriaan

import scala.collection.JavaConverters._

val builder = new KStreamBuilder
val source: KStream[String, String] = builder.stream("streams-file-input")
val counts: KTable[String, java.lang.Long] = source.
  flatMapValues[String] { _.toLowerCase(Locale.getDefault).split(" ").toList.asJava }.
  map[String, String] { (key, value) => new KeyValue(value, value) }.
  groupByKey.count("Counts")

// need to override value serde to Long type
counts.to(Serdes.String, Serdes.Long, "streams-wordcount-output")

--
You received this message because you are subscribed to the Google Groups "scala-user" group.
To unsubscribe from this group and stop receiving emails from it, send an email to scala-user+...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.