some msg lose with error buffered messages send to kafka err: not found topic

[dean-river]

i use-resty-kafka-lua two recive http request and upload msg to kafka, i can recive some msg in kafka, but not all. some is lost. nginx error.log report some error like bellow (i change nginx error log level to info) , when i use logstash http input plugin and kafka output plugin there is no msg loss. i don't know why!

2021/05/13 16:59:27 [error] 41485#0: *134288 [lua] producer.lua:272: buffered messages send to kafka err: not found topic, retryable: true, topic: dp_live_push_origin_quality_raw, partition_id: -1, length: 1, context: ngx.timer, client: 10.71.52.217, server: 0.0.0.0:8888

this is the way i use lua-resty-kafka, every request will use send_msg, dp_live_mcu_push_origin_quality_raw and dp_live_push_origin_quality_raw topic all have 300 brokers

` local _M = {} _M.topics_broker_list = { test = { { host = "127.0.0.1", port = 9092 }, },

dp_live_mcu_push_origin_quality_raw = {
    { host = "dataarch-kafka-h1.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h2.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h3.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h4.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h5.idczw.hb1.kwaidc.com", port = 9092},
},

dp_live_push_origin_quality_raw = {
    { host = "dataarch-kafka-h1.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h2.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h3.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h4.idczw.hb1.kwaidc.com", port = 9092},
    { host = "dataarch-kafka-h5.idczw.hb1.kwaidc.com", port = 9092},
},

}

_M.log_type_2_topic = { test = "test", mcu = "dp_live_mcu_push_origin_quality_raw", srs = "dp_live_push_origin_quality_raw", }

_M.log_type_2_log_id = { test = "test", mcu = "logUploadKafkaTime", srs = "logUploadKafkaTime", }

-- return 0 ok -- return 1 data error -- return 2 server error function _M.send_msg(json_data) local cjson = require "cjson" local producer = require "resty.kafka.producer" local log_type = json_data["@log_type"]

if not log_type then
    return 1, "@log_type not exits"
end

local topic = _M.log_type_2_topic[log_type]
if not topic then
    return 1, "@log_type not config"
end

local broker_list = _M.topics_broker_list[topic]
local bp = producer:new(broker_list, { producer_type = "async" })
local json_str = cjson.encode(json_data)
local ok, err = bp:send(topic, nil, json_str)

if not ok then
    ngx.log(ngx.ERR, "send err=", err, "msg=", json_str)
    return 2, "send kafka error"
end

return 0, "ok"

end

`