Kafka流过滤问题

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())

错误：(33, 25) 缺少扩展函数的参数类型 ((x$1) => x$1.toUpperCase()) textLines.mapValues(_.toUpperCase())

如果我将光标悬停在代码上会出现错误：

类型不匹配，预期：ValueMapper[_ >: String, _ <: notinferedvr> Any 无法将符号解析为大写

我的 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"

我真的不知道如何继续，因为我对 Scala 还很陌生。我想知道是什么问题以及如何解决。

Answer 1

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

这个问题的根本原因是 Scala-Java 互操作性——Kafka Streams API 是用 Java 实现的，但您的应用程序是用 Scala 编写的。值得注意的是，这个问题是由 Java 和 Scala 的类型系统如何交互引起的。例如，Java 中的通用通配符通常会导致此类 Scala 问题。

要解决此问题，您需要在 Scala 应用程序中显式声明类型才能编译代码。例如，您可能需要将将多个 DSL 操作链接成多个语句的单个语句分解，其中每个语句显式声明各自的返回类型。 StreamToTableJoinScalaIntegrationTest 演示了如何显式声明返回变量的类型。

更新

Kafka 2.0（将于 6 月发布）包含一个适当的 Scala API，可以避免这些问题。比较https://cwiki.apache.org/confluence/display/KAFKA/KIP-270+-+A+Scala+Wrapper+Library+for+Kafka+Streams