Apache APISIX中PubSub/Kafka响应Protobuf解码错误问题分析

问题背景

在Apache APISIX 3.6.0版本中，当使用PubSub功能连接Kafka代理时，用户报告了一个Protobuf解码错误。具体表现为：虽然能够成功创建路由并通过WebSocket连接，也能正确从Kafka获取消息，但在尝试发送响应时会出现"type 'PubSubResp' does not exists"的错误。

问题现象

当用户按照以下流程操作时会出现问题：

1. 创建连接到Kafka Broker的路由
2. 通过WebSocket订阅该路由
3. 向带有分区和偏移量的主题发送PubSubReq请求
4. APISIX成功接收Kafka消息后，send_response函数无法正确编码响应消息

错误日志显示："failed to encode response message, err: bad argument #1 to '?' (type 'PubSubResp' does not exists)"

技术分析

这个问题本质上是一个Protobuf状态管理问题。在APISIX的pubsub.lua实现中，当尝试编码响应消息时，Protobuf的状态(pb.state)在初始化和发送响应之间丢失了。

Protobuf在Lua中的实现需要维护一个状态来跟踪已加载的消息类型。在当前的实现中，虽然PubSubResp类型确实定义在pubsub.proto文件中，但由于状态丢失，编码器无法识别这个类型。

解决方案

正确的做法是在发送响应前保存和恢复Protobuf的状态。具体修复方法是在send_resp函数开始时添加以下代码：

local pb_old_state = pb.state(pb_state)

这个修复确保了：

1. 在编码响应时Protobuf的状态与初始化时一致
2. PubSubResp类型能够被正确识别
3. 响应消息能够被成功编码和发送

问题影响

这个bug影响了所有使用APISIX PubSub功能与Kafka集成的用户，特别是在需要接收Kafka消息并返回响应的场景下。虽然消息能够被正确获取，但由于响应编码失败，客户端无法得到预期的结果。

最佳实践

对于使用APISIX与消息队列集成的开发者，建议：

1. 确保使用修复后的版本
2. 在自定义插件开发时注意Protobuf状态管理
3. 对于关键业务场景，充分测试消息收发流程
4. 监控日志中的Protobuf编解码错误

总结

这个案例展示了在Lua中使用Protobuf时状态管理的重要性。虽然Protobuf提供了强大的序列化能力，但在多阶段处理过程中需要特别注意状态的维护。APISIX团队已经通过简单的状态保存修复了这个问题，确保了PubSub功能的完整性和可靠性。