- 功能说明/问题现象: 修改kafka发送逻辑

- 代码提交人: ultrakid
- 自测报告:
- 软件开发号:
- 来源项目编号/版本简称:
- 问题分类:
- 方案/解决办法:
- 有无兼容性问题:
- 有无配合更新模块:
- 需求路径:
- BUG路径:
- 设计文档路径:
- 建议:
- 代码审核人:

src/emqx_plugin_kafka.erl

@@ -332,7 +332,8 @@ produce_kafka_payload(Key, Message) ->
   {ok, MessageBody} = emqx_json:safe_encode(Message),
   % ?LOG_INFO("[KAFKA PLUGIN]Message = ~s~n",[MessageBody]),
   Payload = iolist_to_binary(MessageBody),
-  {ok, _} = brod:produce(emqx_repost_worker, Topic, 0, Key, Payload).
+  brod:produce_cb(emqx_repost_worker, Topic, hash, Key, Payload, fun(_,_) -> ok end),
+  ok.
 
 ntoa({0, 0, 0, 0, 0, 16#ffff, AB, CD}) ->
   inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256});