bug: PubSub/Kafka protobuf decode error when sending response

[arnauddeman]

Current Behavior

I use pubsub to connect apisix to kafka. I succeeded to create a route, to connect to it via websocket and to send a PubSubReq query. The messages are correctly fetched from kafka but then this error is thrown when apisix try to send the response: bad argument #​1 to '?' (type 'PubSubResp' does not exists)

I believe this is a pb state problem and adding this instruction
local pb_old_state = pb.state(pb_state)
at the beginning of send_resp in pubsub.lua fixed the issue for me.

I am not very familiar with protobuf so I am not sure if this instruction is actually missing or if there is an error on my side.

Expected Behavior

Response should be encoded successfully as the type PubSubResp is defined in pubsub.proto

Error Logs

send_resp(): failed to encode response message, err: bad argument #​1 to '?' (type 'PubSubResp' does not exists), client: 172.22.0.1, server: _, request: "GET /kafka HTTP/1.1", host: "localhost:9080"

Steps to Reproduce

1 - Create a route to connect to KAFKA Broker.
2 - Subscribe to this route via web socket.
3 - Send a PubSubReq to a topic with partition and offset in order that Apisix recieve Kafka messages.
4 - The send_response function should fail to encode the response message.

Environment

APISIX docker version, image apache/apisix:3.6.0-debian

[shreemaan-abhishek]

I would like to try to repro this issue but I am not much familiar with kafka, so could you please help me with a self contained example?

[arnauddeman]

I can create a Github project with docker-compose files and some node scripts.
Would it be OK for you ?

[shreemaan-abhishek]

sure

[arnauddeman]

The repository is: https://github.com/arnauddeman/APISIX-Kafka
The explanations are in the Readme file, but let me know if it is not clear or if there is an issue.
I use Linux, you may have to adapt the npm scripts if your OS is not unix based.

[shreemaan-abhishek]

I could reproduce this bug. The reason why this bug occurs is that the pb.state is lost in between init and sent_resp. I have raised this PR to fix the bug: #11135