有没有办法在 Scala 的 Kafka Streams DSL 中命名处理器答案

【问题标题】：Is there a way to name a Processor in Kafka Streams DSL in Scala有没有办法在 Scala 的 Kafka Streams DSL 中命名处理器
【发布时间】：2022-02-23 01:05:10
【问题描述】：

我一直在尝试使用 Scala 中的 Kafka Streams DSL 命名 KStream，但我找不到在 org.apache.kafka.streams.scala.kstream.Consumed 中命名处理器的方法。虽然有一个java方法org.apache.kafka.streams.kstream.Consumed#as但是它抛出了一个异常。有谁知道可以做什么？

ClassCastException invoking Processor. Do the Processor's input types match the deserialized types? Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. Make sure the Processor can accept the deserialized input of type key: [B, and value: [B.
Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes.
org.apache.kafka.streams.errors.StreamsException: ClassCastException invoking Processor. Do the Processor's input types match the deserialized types? Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. Make sure the Processor can accept the deserialized input of type key: [B, and value: [B.
Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes.
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:236)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:216)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:168)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:85)
    at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:678)
    at ...

编辑：我使用的代码：

val someEvents = builder
  .stream[String, String]("some_events")
(org.apache.kafka.streams.kstream.Consumed.as("some_event_stream"))

我应该使用的代码：

val someEvents = builder
  .stream[String, String]("some_events")
(org.apache.kafka.streams.kstream.Consumed.as("some_event_stream")
  .withKeySerde(Serdes.String())
  .withValueSerde(Serdes.String()))

【问题讨论】：

错误与名称无关（您对命名对象执行此操作。错误是指您的 Serde 对象。您能否显示更多代码？
哦，你是对的，问题是我使用java方法Consumed#as没有明确提供Serdes。
那么，你解决了这个问题吗？如果是这样，请将答案放在下面而不是编辑问题

标签： scala apache-kafka apache-kafka-streams

【解决方案1】：

初始代码：

val someEvents = builder
    .stream[String, String]("some_events")
  (org.apache.kafka.streams.kstream.Consumed.as("some_event_stream"))

我犯了两个错误：

将Consumed 参数放在单独的行中。（todo 需要解释为什么它在 Scala 中不起作用）
使用了 Java 方法 org.apache.kafka.streams.kstream.Consumed.as，但没有提供 SerDes。

这行得通：

val someEvents = builder.stream[String, String]("some_events")(org.apache.kafka.streams.kstream.Consumed.as("some_event_stream").withKeySerde(Serdes.String()).withValueSerde(Serdes.String()))

【讨论】：

代码解释有助于其他用户理解差异。