为什么会出现这样的编译错误：“找不到kstream.consumered的隐式值”，我该如何修复它？

Question

我们有这些依赖性：

libraryDependencies += "org.apache.kafka"       %% "kafka-streams-scala"         % kafkaVersion
libraryDependencies += "io.confluent"           % "kafka-streams-avro-serde"     % confluentVersion
libraryDependencies += "io.confluent"           % "kafka-schema-registry-client" % confluentVersion
libraryDependencies += "ch.qos.logback"         % "logback-classic"              % "1.2.3"
libraryDependencies += "com.typesafe"           % "config"                       % "1.4.0"
libraryDependencies += "com.sksamuel.avro4s"    %% "avro4s-core"                 % "3.0.4"

我们使用代码生成器从AVRO模式文件生成Scala案例类。一个这样生成的case类有一个值，作为其字段之一。在AVRO模式中，这是用type=[t1，t2]表示的，因此生成似乎是合理的，即sum类型：可以是t1类型，也可以是t2类型。

问题在于，从主题到案例类（二进制）的反序列化路径缺少什么-

基本上我得到这个错误目前：

could not find implicit value for parameter consumed: org.apache.kafka.streams.scala.kstream.Consumed[String, custom.UserEvent]
[error]       .stream[String, UserEvent]("schma.avsc")

第一个想法是kafka-stream-avro-serde，但可能这个库只确保AVRO Map的Serde[GenericRecer]，而不是case类。因此，其他依赖项之一是帮助AVRO GenericRecer将case类映射和返回。我们还有一些手写的代码，可以从模式中生成case类，这似乎可以直接用于喷涂json。

我在想（二进制）

我现在正在尝试创建一个Serde[用户事件]实例。因此，在我的理解中，这将涉及在用户事件和AVRO通用记录之间转换，类似于地图，然后在AVRO记录和二进制之间转换——这可能包括在kafka-stream-avro-serde依赖关系中，就像应该有一个Serde[通用记录]或类似的东西一样。

导入明智的，我们有这个导入含义：

import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.streams.Topology
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala.Serdes
import org.apache.kafka.streams.scala.Serdes._
import org.apache.kafka.streams.scala.kstream.Consumed

Answer 1

对我来说，我必须更好地遵循指示，并添加一个隐式的serde实现。他们在链接中的示例如下所示：

// An implicit Serde implementation for the values we want to
// serialize as avro
implicit val userClicksSerde: Serde[UserClicks] = new AvroSerde

有关更完整的示例，请参见其avro库的scala测试：

    // Make an implicit serde available for GenericRecord, which is required for operations such as `to()` below.
    implicit val genericAvroSerde: Serde[GenericRecord] = {
      val gas = new GenericAvroSerde
      val isKeySerde: Boolean = false
      gas.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, cluster.schemaRegistryUrl), isKeySerde)
      gas
    }

Answer 2

你导入了相应的软件包吗？

import org.apache.kafka.streams.scala.ImplicitConversions._

查阅https://kafka.apache.org/24/documentation/streams/developer-guide/dsl-api.html#scala-数字用户线

Answer 3

事实上，缺少一个导入。现在它可以编译了。下面是导入：

import org.apache.kafka.streams.Topology
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala.Serdes._