我试图在 Scala 中实现一个简单的流处理应用程序，但遇到类型转换问题。任何指针高度赞赏

这是我试图让 Scala 编译编译的代码片段

case class ItemValue(LOCATION:String,
                          U_ID:String,
                          UOM:String,
                          R_ID:String,
                          ITEM_TYPE:String,
                          B_ID:String,
                          RECORDED_TIMESTAMP:String,
                          P_ID:String,
                          VALUE:String,
                          RECORDED_DTM:String,
                          DATA_TYPE:String)

val valueSerde : JSONSerde[ItemValue] = new JSONSerde[ItemValue]
  val consumed = Consumed.`with`(Serdes.String(), valueSerde)
  val items: KStream[String, ItemValue] = builder.stream("raw_topic", consumed)
    .groupBy(
      new KeyValueMapper[String, ItemValue, KeyValue[String, ItemValue]] {
        override def apply(key: String, value: ItemValue) : KeyValue[String, ItemValue] = {
          new KeyValue(value.P_ID+"_"+value.U_ID+"_"+value.R_ID+"_"+value.B_ID+"_"+value.ITEM_TYPE,value)
        }
      },
      Serialized.`with`(
        Serdes.String(), 
        new JSONSerde[ItemValue])
    )

这是我无法理解的编译器错误

type mismatch;
 found   : org.apache.kafka.streams.kstream.KeyValueMapper[String,com.example.streams.ItemValue,org.apache.kafka.streams.KeyValue[String,com.example.streams.ItemValue]]
 required: org.apache.kafka.streams.kstream.KeyValueMapper[_ >: String, _ >: com.example.streams.ItemValue, Object]
Note: String <: Any, but Java-defined trait KeyValueMapper is invariant in type K.
You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
Note: com.example.streams.ItemValue <: Any, but Java-defined trait KeyValueMapper is invariant in type V.
You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
Note: org.apache.kafka.streams.KeyValue[String,com.example.streams.ItemValue] <: Object, but Java-defined trait KeyValueMapper is invariant in type VR.
You may wish to investigate a wildcard type such as `_ <: Object`. (SLS 3.2.10)
      new KeyValueMapper[String, ItemValue, KeyValue[String, ItemValue]] {
      ^
ItemStatsApp.scala:59: type mismatch;
 found   : org.apache.kafka.streams.kstream.Serialized[String,com.example.streams.ItemValue]
 required: org.apache.kafka.streams.kstream.Serialized[Object,com.example.streams.ItemValue]
Note: String <: Object, but Java-defined class Serialized is invariant in type K.
You may wish to investigate a wildcard type such as `_ <: Object`. (SLS 3.2.10)
      Serialized.`with`(
                       ^
two errors found

Kafka 常见问题解答(来自 Confluent)和答案这里说如果在 Scala 中没有明确定义类型，可能会发生这种情况，但是我无法看到我没有明确的地方。

scala apache-kafka