Kafka Streams转换后的数据无法通过JDBC Sink写入数据库求助

核心问题 #

你用kafka-json-schema-console-producer能正常写入，是因为这个工具会自动给JSON数据加上Schema Registry的元数据（比如schema ID），输出的是Confluent标准的带Schema包装的JSON格式。但你的Kafka Streams程序默认输出的是纯JSON字符串，JDBC Sink Connector如果配置了依赖Schema Registry解析，就识别不了这种无元数据的消息。

具体调整步骤 #

1. 让Kafka Streams输出带Schema元数据的消息 #

修改Streams代码，改用Confluent的JsonSchemaSerde序列化输出数据，替代默认的字符串/普通JSON序列化。

• 先添加依赖（Maven示例）：

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-streams-json-schema-serde</artifactId>
    <version>${confluent.version}</version>
</dependency>

• 配置Serde并绑定到输出主题：

// 配置Schema Registry地址
Map<String, String> serdeConfigs = new HashMap<>();
serdeConfigs.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://你的SchemaRegistry地址:8081");

// 用自定义POJO的情况（假设你的数据对应YourData类）
JsonSchemaSerde<YourData> jsonSchemaSerde = new JsonSchemaSerde<>(YourData.class);
jsonSchemaSerde.configure(serdeConfigs, false); // false表示序列化值（不是键）

// 数据流处理并输出
streamsBuilder.stream("topicstream")
    .mapValues(originalData -> {
        // 你的转换逻辑，返回YourData实例
        return transformedData;
    })
    .to("jdbcsinktopic2", Produced.with(Serdes.String(), jsonSchemaSerde));

如果不用POJO，用GenericRecord动态处理Schema也可以：

JsonSchemaSerde<GenericRecord> jsonSchemaSerde = new JsonSchemaSerde<>();
jsonSchemaSerde.configure(serdeConfigs, false);

2. 检查JDBC Sink Connector的配置 #

确保Sink的转换器配置正确，要和Schema Registry对接：

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.json.JsonSchemaConverter
value.converter.schema.registry.url=http://你的SchemaRegistry地址:8081
value.converter.schemas.enable=true

关键是value.converter要用JsonSchemaConverter，并指定Schema Registry地址，开启schema支持。

3. 验证输出格式 #

启动Streams程序后，用命令行查看jdbcsinktopic2的消息：

kafka-console-consumer --bootstrap-server 你的Kafka地址:9092 --topic jdbcsinktopic2 --from-beginning

如果输出是类似下面的格式，说明配置生效了：

{"schemaId": 1, "payload": {"xyz": "Test from couchbase"}}

要是还是纯JSON，说明Serde没配置对，回去检查代码里的序列化器绑定。

4. 确认Schema匹配 #

确保Streams程序使用的Schema和你在Schema Registry里配置的完全一致（字段名、类型都要对应）。如果Streams自动注册Schema，要检查注册的Schema是否和MySQL表结构兼容（比如字段类型要匹配）。

备选方案（不用Schema Registry） #

如果不想用Schema Registry，也可以让JDBC Sink直接解析纯JSON，修改Sink配置：

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
# 删掉value.converter.schema.registry.url这个参数

这种方式要求JSON字段名和MySQL表列名完全一致，且类型兼容，适合简单场景。

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