- 功能说明/问题现象: 将kafka的依赖库从ekaf替换为brod

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

etc/emqx_plugin_kafka.conf

@@ -1,31 +0,0 @@
-##--------------------------------------------------------------------
-## kafka Bridge
-##--------------------------------------------------------------------
-
-## The Kafka loadbalancer node host that bridge is listening on.
-##
-## Value: 127.0.0.1, localhost
-###kafka.host = 192.168.67.16
-kafka.host = 127.0.0.1
-
-## The kafka loadbalancer node port that bridge is listening on.
-##
-## Value: Port
-###kafka.port = 9092
-kafka.port = 9092
-
-## The kafka loadbalancer node partition strategy.
-##
-## Value: random, sticky_round_robin, strict_round_robin, custom
-kafka.partitionstrategy = strict_round_robin
-
-## Each worker represents a connection to a broker + topic + partition combination.
-## You can decide how many workers to start for each partition.
-##
-## Value: 
-kafka.partitionworkers = 8
-
-## payload topic.
-##
-## Value: string
-kafka.payloadtopic = emqx-topic

etc/emqx_plugin_kafka.config

@@ -0,0 +1,10 @@
+[
+    {emqx_plugin_kafka, [
+        {kafka_address_list, [{"localhost", 9092}]},    %% kafka地址，可配置多个
+        {kafka_config, [                                %% 这部分是https://github.com/kafka4beam/brod库需要的配置
+            {reconnect_cool_down_seconds, 10},          %% socket error recovery
+            {query_api_versions, true}                  %% Kafka 0.9.x or Earlier请设置为false
+        ]},
+        {topic, "emqx-topic"}                           %% 转发的目标topic
+    ]}
+].

priv/emqx_plugin_kafka.schema

@@ -1,35 +0,0 @@
-%%-*- mode: erlang -*-
-
-{mapping, "kafka.host", "emqx_plugin_kafka.broker", [
-  {default, "192.168.0.128"},
-  {datatype, string}
-]}.
-
-{mapping, "kafka.port", "emqx_plugin_kafka.broker", [
-  {default, "9092"},
-  {datatype, string}
-]}.
-
-{mapping, "kafka.partitionstrategy", "emqx_plugin_kafka.broker", [
-  {default, "strict_round_robin"},
-  {datatype, string}
-]}.
-
-{mapping, "kafka.partitionworkers", "emqx_plugin_kafka.broker", [
-  {default, 8},
-  {datatype, integer}
-]}.
-
-{mapping, "kafka.payloadtopic", "emqx_plugin_kafka.broker", [
-  {default, "Processing"},
-  {datatype, string}
-]}.
-
-{translation, "emqx_plugin_kafka.broker", fun(Conf) ->
-  KafkaHost = cuttlefish:conf_get("kafka.host", Conf),
-  KafkaPort = cuttlefish:conf_get("kafka.port", Conf),
-  KafkaPartitionStrategy = cuttlefish:conf_get("kafka.partitionstrategy", Conf),
-  KafkaPartitionWorkers = cuttlefish:conf_get("kafka.partitionworkers", Conf),
-  KafkaPayloadTopic = cuttlefish:conf_get("kafka.payloadtopic", Conf),
-  [{host, KafkaHost}, {port, KafkaPort}, {partitionstrategy, KafkaPartitionStrategy}, {partitionworkers, KafkaPartitionWorkers}, {payloadtopic, KafkaPayloadTopic}]
-  end}.

rebar.config

@@ -1,6 +1,6 @@
 {deps, [
     {cuttlefish, {git, "https://github.com/emqx/cuttlefish.git",  {branch, "main"}}},
-    {ekaf, {git, "https://github.com/helpshift/ekaf.git",  {branch, "master"}}},
+    {brod, {git, "https://github.com/kafka4beam/brod.git",  {tag, "3.16.3"}}},
     {emqx, {git_subdir, "https://github.com/ULTRAKID/emqx", {tag, "v4.3.12-with-kafka"}, "apps/emqx"}}
 ]}.
 

src/emqx_plugin_kafka.app.src

@@ -1,5 +1,5 @@
 {application, emqx_plugin_kafka,
- [{description, "EMQ X plugin kafka bridge, Edit by yzs-lyg, ultrakid"},
+ [{description, "EMQ X plugin kafka bridge, Edit by ultrakid, original coder: yzs-lyg"},
   {vsn, "git"},
   {modules, []},
   {registered, [emqx_plugin_kafka_sup]},
@@ -7,8 +7,8 @@
   {mod, {emqx_plugin_kafka_app,[]}},
   {env, []},
   {licenses, ["Apache-2.0"]},
-  {maintainers, ["lyg <3613840847@qq.com>", "ultrakid <ultrakid@qq.com>"]},
-  {links,[{"Github","https://github.com/yzs/emqx-plugin-kafka"}, {"Github","https://github.com/ULTRAKID/emqx_plugin_kafka"}]}
+  {maintainers, ["ultrakid <ultrakid@qq.com>", "lyg <3613840847@qq.com>"]},
+  {links,[{"Github","https://github.com/ULTRAKID/emqx_plugin_kafka"}, {"Github","https://github.com/yzs/emqx-plugin-kafka"}]}
  ]
 }.
 

src/emqx_plugin_kafka.erl

@@ -58,7 +58,7 @@
 
 %% Called when the plugin application start
 load(Env) ->
-  ekaf_init([Env]),
+  kafka_init([Env]),
   emqx:hook('client.connect', {?MODULE, on_client_connect, [Env]}),
   emqx:hook('client.connack', {?MODULE, on_client_connack, [Env]}),
   emqx:hook('client.connected', {?MODULE, on_client_connected, [Env]}),
@@ -104,10 +104,10 @@ on_client_connected(ClientInfo = #{clientid := ClientId}, ConnInfo, _Env) ->
     {ipaddress, iolist_to_binary(ntoa(IpAddr))},
     {proto_name, maps:get(proto_name, ConnInfo)},
     {proto_ver, maps:get(proto_ver, ConnInfo)},
-    {timestamp, Now},
+    {ts, Now},
     {online, Online}
   ],
-  produce_kafka_payload(Payload),
+  produce_kafka_payload(ClientId, Payload),
   ok.
 
 on_client_disconnected(ClientInfo = #{clientid := ClientId}, ReasonCode, ConnInfo, _Env) ->
@@ -121,10 +121,10 @@ on_client_disconnected(ClientInfo = #{clientid := ClientId}, ReasonCode, ConnInf
     {device_id, ClientId},
     {username, maps:get(username, ClientInfo)},
     {reason, ReasonCode},
-    {timestamp, Now},
+    {ts, Now},
     {online, Online}
   ],
-  produce_kafka_payload(Payload),
+  produce_kafka_payload(ClientId, Payload),
   ok.
 
 on_client_authenticate(_ClientInfo = #{clientid := ClientId}, Result, _Env) ->
@@ -148,9 +148,9 @@ on_client_subscribe(#{clientid := ClientId}, _Properties, TopicFilters, _Env) ->
     {action, Action},
     {topic, Topic},
     {qos, maps:get(qos, Qos)},
-    {timestamp, Now}
+    {ts, Now}
   ],
-  produce_kafka_payload(Payload),
+  produce_kafka_payload(ClientId, Payload),
   ok.
 %%---------------------client subscribe stop----------------------%%
 on_client_unsubscribe(#{clientid := ClientId}, _Properties, TopicFilters, _Env) ->
@@ -162,9 +162,9 @@ on_client_unsubscribe(#{clientid := ClientId}, _Properties, TopicFilters, _Env)
     {device_id, ClientId},
     {action, Action},
     {topic, Topic},
-    {timestamp, Now}
+    {ts, Now}
   ],
-  produce_kafka_payload(Payload),
+  produce_kafka_payload(ClientId, Payload),
   ok.
 
 on_message_dropped(#message{topic = <<"$SYS/", _/binary>>}, _By, _Reason, _Env) ->
@@ -179,8 +179,8 @@ on_message_dropped(Message, _By = #{node := Node}, Reason, _Env) ->
 on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
   ok;
 on_message_publish(Message, _Env) ->
-  {ok, Payload} = format_payload(Message),
-  produce_kafka_payload(Payload),
+  {ok, ClientId, Payload} = format_payload(Message),
+  produce_kafka_payload(ClientId, Payload),
   ok.
 %%---------------------message publish stop----------------------%%
 
@@ -202,7 +202,7 @@ on_message_delivered(_ClientInfo = #{clientid := ClientId}, Message, _Env) ->
     {cluster_node, node()},
     {ts, Timestamp}
   ],
-  produce_kafka_payload(Content),
+  produce_kafka_payload(ClientId, Content),
   ok.
 
 on_message_acked(_ClientInfo = #{clientid := ClientId}, Message, _Env) ->
@@ -223,7 +223,7 @@ on_message_acked(_ClientInfo = #{clientid := ClientId}, Message, _Env) ->
     {cluster_node, node()},
     {ts, Timestamp}
   ],
-  produce_kafka_payload(Content),
+  produce_kafka_payload(ClientId, Content),
   ok.
 
 %%--------------------------------------------------------------------
@@ -260,30 +260,19 @@ on_session_terminated(_ClientInfo = #{clientid := ClientId}, Reason, SessInfo, _
     [ClientId, Reason, SessInfo]),
   ok.
 
-ekaf_init(_Env) ->
+kafka_init(_Env) ->
   io:format("Init emqx plugin kafka....."),
-  {ok, BrokerValues} = application:get_env(emqx_plugin_kafka, broker),
-  KafkaHost = proplists:get_value(host, BrokerValues),
-  ?LOG_INFO("[KAFKA PLUGIN]KafkaHost = ~s~n", [KafkaHost]),
-  KafkaPort = proplists:get_value(port, BrokerValues),
-  ?LOG_INFO("[KAFKA PLUGIN]KafkaPort = ~s~n", [KafkaPort]),
-  KafkaPartitionStrategy = proplists:get_value(partitionstrategy, BrokerValues),
-  KafkaPartitionWorkers = proplists:get_value(partitionworkers, BrokerValues),
-  KafkaTopic = proplists:get_value(payloadtopic, BrokerValues),
+  {ok, AddressList} = application:get_env(emqx_plugin_kafka, kafka_address_list),
+  ?LOG_INFO("[KAFKA PLUGIN]KafkaAddressList = ~s~n", [AddressList]),
+  {ok, KafkaConfig} = application:get_env(emqx_plugin_kafka, kafka_config),
+  ?LOG_INFO("[KAFKA PLUGIN]KafkaConfig = ~s~n", [KafkaConfig]),
+  {ok, KafkaTopic} = application:get_env(emqx_plugin_kafka, topic),
   ?LOG_INFO("[KAFKA PLUGIN]KafkaTopic = ~s~n", [KafkaTopic]),
-  application:set_env(ekaf, ekaf_bootstrap_broker, {KafkaHost, list_to_integer(KafkaPort)}),
-  application:set_env(ekaf, ekaf_partition_strategy, list_to_atom(KafkaPartitionStrategy)),
-  application:set_env(ekaf, ekaf_per_partition_workers, KafkaPartitionWorkers),
-  application:set_env(ekaf, ekaf_bootstrap_topics, list_to_binary(KafkaTopic)),
-  application:set_env(ekaf, ekaf_buffer_ttl, 10),
-  application:set_env(ekaf, ekaf_max_downtime_buffer_size, 5),
-  % {ok, _} = application:ensure_all_started(kafkamocker),
-  {ok, _} = application:ensure_all_started(gproc),
-  % {ok, _} = application:ensure_all_started(ranch),
-  {ok, _} = application:ensure_all_started(ekaf).
-
-ekaf_get_topic() ->
-  {ok, Topic} = application:get_env(ekaf, ekaf_bootstrap_topics),
+  ok = brod:start_client(AddressList, emqx_repost_worker, KafkaConfig),
+  ok = brod:start_producer(emqx_repost_worker, KafkaTopic, []).
+
+get_kafka_topic() ->
+  {ok, Topic} = application:get_env(emqx_plugin_kafka, topic),
   Topic.
 
 
@@ -293,7 +282,7 @@ format_payload(Message) ->
   Tail = string:right(binary_to_list(Topic), 4),
   RawType = string:equal(Tail, <<"_raw">>),
   % ?LOG_INFO("[KAFKA PLUGIN]Tail= ~s , RawType= ~s~n",[Tail,RawType]),
-
+  ClientId = Message#message.from,
   MsgPayload = Message#message.payload,
   % ?LOG_INFO("[KAFKA PLUGIN]MsgPayload : ~s~n", [MsgPayload]),
   if
@@ -304,13 +293,13 @@ format_payload(Message) ->
       MsgPayload64 = MsgPayload
   end,
   Payload = [{action, message_publish},
-    {device_id, Message#message.from},
+    {device_id, ClientId},
     {username, Username},
     {topic, Topic},
     {payload, MsgPayload64},
     {ts, Message#message.timestamp}],
 
-  {ok, Payload}.
+  {ok, ClientId, Payload}.
 
 
 %% Called when the plugin application stop
@@ -335,12 +324,12 @@ unload() ->
   emqx:unhook('message.acked', {?MODULE, on_message_acked}),
   emqx:unhook('message.dropped', {?MODULE, on_message_dropped}).
 
-produce_kafka_payload(Message) ->
-  Topic = ekaf_get_topic(),
+produce_kafka_payload(Key, Message) ->
+  Topic = get_kafka_topic(),
   {ok, MessageBody} = emqx_json:safe_encode(Message),
   % ?LOG_INFO("[KAFKA PLUGIN]Message = ~s~n",[MessageBody]),
   Payload = iolist_to_binary(MessageBody),
-  ekaf:produce_async_batched(Topic, Payload).
+  {ok, _} = brod:produce(emqx_repost_worker, Topic, 0, Key, Payload).
 
 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});