kafka消费端sql注入风险源于将不可信的record.value()等输入直接拼入sql，与kafka本身无关；需严格使用参数化查询，对json解析值、嵌套字段、动态表名等均须白名单校验或占位符绑定，并与kafka offset提交机制协同防御。
Kafka消费端写入数据库时，SQL注入风险和Kafka本身无关，而是源于你把record.value()或record.key()这类不可信输入，直接拼进SQL字符串里——哪怕用了PreparedStatement，只要参数绑定没做对，照样中招。
为什么消费者端容易忽略SQL注入
很多人以为“Kafka消息是内部系统发的，可信”，但实际场景中：消息可能来自第三方对接、前端埋点上报、日志采集管道，甚至被恶意篡改过（比如中间代理、配置错误导致topic混用）。一旦value含单引号、分号、注释符或UNION SELECT片段，又没走参数化，就等于把数据库钥匙交出去。
- 常见错误：用
String.format("INSERT INTO t VALUES ('%s')", record.value())拼SQL - 更隐蔽的错：用ORM但写了原生SQL查询，且用
#{}或${}不加区分（MyBatis里${}直接拼接） - 误判“已过滤”：只对字段做了
trim()或replaceAll("'", "")，但绕过方式极多（如/**/注释、双写编码、空字节）
必须用参数化查询，且绑定位置要严格匹配
不是“用了PreparedStatement就安全”，关键在怎么设参数。下面这些写法都危险：
-
stmt.setString(1, "admin' -- ")→ 安全，参数值被整体转义 -
stmt.executeUpdate("UPDATE t SET name = '" + name + "' WHERE id = 1")→ 危险，字符串拼接 -
String sql = "SELECT * FROM users WHERE status = " + statusValue;→ 危险，数字型也需占位符 -
query("SELECT * FROM t WHERE id IN (" + idsStr + ")")→ 危险，IN子句必须动态生成参数占位符，不能拼列表
正确做法：所有用户控制的字段，无论类型、来源、是否“看起来干净”，一律走?占位符，并调用对应setXxx()方法。
消费逻辑里最容易踩坑的三个地方
你在写processMessage(ConsumerRecord)时，这几个点常被跳过校验：
-
JSON解析后取值：比如
new JSONObject(record.value()).getString("username")，这个username仍是外部输入，不能直接进SQL -
多层嵌套字段：从Kafka消息里解析出
user.profile.phone，再拿去查库——只要路径里任意一环可被控制，就必须参数化 -
动态表名/列名：绝对禁止用消息内容拼
INSERT INTO ${tableName}；如真需动态，必须白名单校验（如Set.of("orders", "logs").contains(tableName)）
配合Kafka消费模型的最小防御组合
光防SQL注入不够，得和Kafka可靠性机制对齐：
- 关闭
enable.auto.commit=true，避免SQL执行失败但offset已提交，导致数据丢失+注入同时发生 - SQL执行失败时，别直接
commitSync()，应抛异常并让Kafka重试（或进DLQ），否则“注入失败”和“业务失败”都掩盖了 - 如果用了
commitAsync()，回调里检查SQL异常是否影响了本次提交——异步提交失败不会阻塞，但可能造成重复消费，进而重复执行恶意SQL（即使参数化，重复插入也可能引发业务问题）
真正难的不是写对一条PreparedStatement，而是在整个消费链路里，所有从record流出的数据，在落到SQL之前，都经过同一套参数化规则——这需要代码审查习惯，而不是某个开关能解决。
