您必须包含您正在使用的两个类的Serdes。存在用于字符串的Serdes，但需要为JsonNode创建自定义Serdes。在此之后，您可以使用此代码。

KStream<String, Record> inputStream = streamsBuilder
            .stream(topic, Consumed.with(Serdes.String(), CustomSerdes.JsonNode()));

可用于SERDES的代码为：

public class CustomSerdes {
  private CustomSerdes() {}

  public static Serde<JsonNode> JsonNode() {
    JsonSerializer<JsonNode> serializer = new JsonSerializer<>();
    JsonDeserializer<JsonNode> deserializer = new JsonDeserializer<>(JsonNode.class);
    return Serdes.serdeFrom(serializer, deserializer);
}