如何将Kafka Topic中的字符串转换为KStream<String, JsonNode>？

你的代码无法运行的核心原因是：Kafka Streams默认使用StringDeserializer来反序列化消息值，而你需要将值反序列化为JsonNode类型，必须显式配置对应的JSON反序列化器。

步骤1：确保依赖正确 #

如果使用Maven，确保引入Kafka Streams和Jackson相关依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>你的Kafka版本</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.x.x</version>
</dependency>

步骤2：配置Kafka Streams的反序列化器 #

你有两种配置方式：

方式一：全局配置（所有流都使用该反序列化器）

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "json-consumer-app");
streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

// 配置键的反序列化器为String类型
streamsProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 配置值的反序列化器为JSON类型，目标类是JsonNode
streamsProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());
streamsProps.put(JsonDeserializer.VALUE_TYPE_CLASS_CONFIG, JsonNode.class.getName());
// 禁用类型头（如果你的消息没有携带类型信息头）
streamsProps.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, "false");

StreamsBuilder builder = new StreamsBuilder();
// 此时直接构建流即可得到KStream<String, JsonNode>
KStream<String, JsonNode> inputStream = builder.stream("你的topic名称");

方式二：针对单个流配置（仅当前流使用该反序列化器）

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "json-consumer-app");
streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

StreamsBuilder builder = new StreamsBuilder();
// 显式指定键值的Serde，Serdes.Json()会自动使用JsonDeserializer/JsonSerializer
KStream<String, JsonNode> inputStream = builder.stream(
    "你的topic名称",
    Consumed.with(Serdes.String(), Serdes.Json(JsonNode.class))
);

额外注意事项 #

你发送到Kafka的字符串{{"header" : "K"},{"body" : "Sghd"}}是无效JSON格式，正确的JSON数组应该是[{"header":"K"},{"body":"Sghd"}]，如果是单个对象则需保证外层只有一对大括号。格式错误会直接导致反序列化失败，务必先修正消息的JSON格式。

内容的提问来源于stack exchange，提问作者Rahul