Kafka Streams序列化JSONObject时生成空Schema的问题求助

你遇到的核心问题是：KafkaJsonSchemaSerde 依赖Jackson的JSON Schema生成器，但org.json.JSONObject并非Jackson原生类型，其对应的Jackson序列化器未实现Schema生成所需的acceptJsonFormatVisitor接口，导致无法自动生成有效Schema，最终输出空Schema结构。

方法一：改用Jackson原生JsonNode类型（推荐） #

JsonNode是Jackson的原生JSON节点类型，KafkaJsonSchemaSerde可以自动识别其结构并生成正确的Schema，同时更适配Kafka Streams的处理逻辑。

代码示例 #

// 初始化输入Serde，使用JsonNode替代JSONObject
KafkaJsonSchemaSerde<JsonNode> inputSerde = new KafkaJsonSchemaSerde<>(JsonNode.class);
Map<String, Object> baseConf = new HashMap<>();
baseConf.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
inputSerde.configure(baseConf, false);

// 读取输入流
KStream<String, JsonNode> inputStream = builder.stream(inputTopic, Consumed.with(Serdes.String(), inputSerde));

// 处理流：移除非原始类型字段
KStream<String, JsonNode> primitiveStream = inputStream.mapValues(value -> {
    ObjectNode objectNode = (ObjectNode) value;
    List<String> fieldsToRemove = new ArrayList<>();
    
    // 遍历所有字段，标记需要移除的对象/数组类型字段
    objectNode.fieldNames().forEachRemaining(key -> {
        JsonNode node = objectNode.get(key);
        if (node.isObject() || node.isArray()) {
            logger.info("Key: '{}' removed, is an object or array.", key);
            fieldsToRemove.add(key);
        }
    });
    
    // 批量移除字段
    fieldsToRemove.forEach(objectNode::remove);
    return objectNode;
});

// 初始化输出Serde，直接使用JsonNode类型
KafkaJsonSchemaSerde<JsonNode> outputSerde = new KafkaJsonSchemaSerde<>(JsonNode.class);
outputSerde.configure(baseConf, false);

// 写入目标主题
primitiveStream.to(inputTopic + ".primitive", Produced.with(Serdes.String(), outputSerde));

方法二：手动注册Schema并关闭自动生成 #

如果必须保留JSONObject类型，可以手动创建匹配处理后数据结构的Schema，提前注册到Schema Registry，再配置Serde使用指定Schema。

代码示例 #

// 1. 手动定义处理后数据的JSON Schema（根据实际字段调整）
String targetSchemaStr = """
{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "title": "PrimitiveData",
  "properties": {
    "id": {"type": "integer"},
    "username": {"type": "string"},
    "create_time": {"type": "string", "format": "date-time"}
  },
  "additionalProperties": true
}
""";

// 2. 将Schema注册到Schema Registry
SchemaRegistryClient registryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 100);
String outputSubject = inputTopic + ".primitive-value";
JsonSchema targetSchema = JsonSchemaFactory.getInstance(SchemaVersion.DRAFT_07).getSchema(targetSchemaStr);
int schemaId = registryClient.register(outputSubject, targetSchema);

// 3. 配置Serde关闭自动注册，指定已注册的Schema ID
KafkaJsonSchemaSerde<JSONObject> jsonObjSerde = new KafkaJsonSchemaSerde<>(JSONObject.class);
Map<String, Object> serdeConf = new HashMap<>();
serdeConf.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
serdeConf.put(AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false);
serdeConf.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_ID_CONFIG, schemaId);
jsonObjSerde.configure(serdeConf, false);

// 写入目标主题
primitiveStream.to(inputTopic + ".primitive", Produced.with(Serdes.String(), jsonObjSerde));

• 使用JsonNode时，无需手动维护Schema，Serde会根据实际数据结构自动生成并注册，更适合动态字段场景。
• 手动注册Schema时，必须确保处理后的JSONObject结构与预定义Schema完全匹配，否则后续JDBC Sink连接器可能出现数据写入错误。
• 关闭自动注册后，需保证Schema已提前存在于Registry中，否则会触发"Schema未找到"的异常。

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