問題與卡夫卡流過濾

Question

我試圖運行從下面的例子中一個基本的應用程序：

https://github.com/confluentinc/examples/blob/3.3.x/kafka-streams/src/main/scala/io/confluent/examples/streams/MapFunctionScalaExample.scala

但是我發現了一個例外，在這一行：

// Variant 1: using `mapValues` 
val uppercasedWithMapValues: KStream[Array[Byte], String] = textLines.mapValues(_.toUpperCase()) 

Error:(33, 25) missing parameter type for expanded function ((x$1) => x$1.toUpperCase()) textLines.mapValues(_.toUpperCase())

如果我將鼠標懸停在代碼上，將出現錯誤：

Type mismatch, expected: ValueMapper[_ >: String, _ <: NotInferedVR], actual: (Any) => Any Cannot resolve symbol toUpperCase

項

內容我SBT文件：

name := "untitled1" 

version := "0.1" 

scalaVersion := "2.11.11" 

// https://mvnrepository.com/artifact/org.apache.kafka/kafka_2.11 
libraryDependencies += "org.apache.kafka" % "kafka_2.11" % "0.11.0.0" 

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients 
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.11.0.0" 

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams 
libraryDependencies += "org.apache.kafka" % "kafka-streams" % "0.11.0.0" 

// https://mvnrepository.com/artifact/org.apache.kafka/connect-api 
libraryDependencies += "org.apache.kafka" % "connect-api" % "0.11.0.0" 

我真的不知道如何與繼續因爲我很新的斯卡拉。我想知道有什麼問題以及如何解決問題。

Answer 1

從http://docs.confluent.io/current/streams/faq.html#scala-compile-error-no-type-parameter-java-defined-trait-is-invariant-in-type-t

The root cause of this problem is Scala-Java interoperability – the Kafka Streams API is implemented in Java, but your application is written in Scala. Notably, this problem is caused by how the type systems of Java and Scala interact. Generic wildcards in Java, for example, are often causing such Scala issues.

To fix the problem you would need to declare types explicitly in your Scala application in order for the code to compile. For example, you may need to break a single statement that chains multiple DSL operations into multiple statements, where each statement explicitly declares the respective return types. The StreamToTableJoinScalaIntegrationTest demonstrates how the types of return variables are explicitly declared.