适配emqx v4.3版本

rebar.config

@@ -1,7 +1,7 @@
 {deps, [
     {cuttlefish, {git, "https://github.com/emqx/cuttlefish.git",  {branch, "main"}}},
-    {ekaf, {git, "https://github.com/helpshift/ekaf.git",  {branch, "master"}}}
-    %% {emqx, {git, "https://github.com/emqx/emqx.git",  {branch, "main-v4.3"}}}
+    {ekaf, {git, "https://github.com/helpshift/ekaf.git",  {branch, "master"}}},
+    {emqx, {git, "https://github.com/ULTRAKID/emqx",  {tag, "v4.3.12-with-kafka"}}}
 ]}.
 
 {erl_opts, [debug_info]}.

rebar.config.script

@@ -44,10 +44,11 @@ GitDescribe = case re:run(Branch, "master|^dev/|^hotfix/", [{capture, none}]) of
 
 UrlPrefix = "https://github.com/emqx/",
 
-EMQX_DEP = {emqx, {git, UrlPrefix ++ "emqx", GitDescribe}},
+%% EMQX_DEP = {emqx, {git, UrlPrefix ++ "emqx", GitDescribe}},
 EMQX_MGMT_DEP = {emqx_management, {git, UrlPrefix ++ "emqx-management", GitDescribe}},
 
-NewDeps = [EMQX_DEP, EMQX_MGMT_DEP | DEPS],
+%% NewDeps = [EMQX_DEP, EMQX_MGMT_DEP | DEPS],
+NewDeps = [EMQX_MGMT_DEP | DEPS],
 
 CONFIG1 = lists:keystore(deps, 1, CONFIG, {deps, NewDeps}),
 

src/emqx_plugin_kafka.erl

@@ -22,8 +22,8 @@
 
 % -include_lib("emqx/include/emqx.hrl").
 
--include_lib("emqx.hrl").
--include_lib("logger.hrl").
+-include("emqx.hrl").
+-include("logger.hrl").
 
 -export([load/1, unload/0]).
 
@@ -128,12 +128,12 @@ on_client_disconnected(ClientInfo = #{clientid := ClientId}, ReasonCode, ConnInf
   ok.
 
 on_client_authenticate(_ClientInfo = #{clientid := ClientId}, Result, _Env) ->
-%%  ?INFO("[KAFKA PLUGIN]Client(~s) authenticate, Result:~n~p~n", [ClientId, Result]),
+  ?INFO("[KAFKA PLUGIN]Client(~s) authenticate, Result:~n~p~n", [ClientId, Result]),
   ok.
 
 on_client_check_acl(_ClientInfo = #{clientid := ClientId}, Topic, PubSub, Result, _Env) ->
-%%  ?INFO("[KAFKA PLUGIN]Client(~s) check_acl, PubSub:~p, Topic:~p, Result:~p~n",
-%%    [ClientId, PubSub, Topic, Result]),
+  ?INFO("[KAFKA PLUGIN]Client(~s) check_acl, PubSub:~p, Topic:~p, Result:~p~n",
+    [ClientId, PubSub, Topic, Result]),
   ok.
 
 %%---------------------------client subscribe start--------------------------%%
@@ -232,7 +232,7 @@ on_message_acked(_ClientInfo = #{clientid := ClientId}, Message, _Env) ->
 %%--------------------------------------------------------------------
 
 on_session_created(#{clientid := ClientId}, SessInfo, _Env) ->
-  %?INFO("[KAFKA PLUGIN]Session(~s) created, Session Info:~n~p~n", [ClientId, SessInfo]).
+  ?INFO("[KAFKA PLUGIN]Session(~s) created, Session Info:~n~p~n", [ClientId, SessInfo]).
   ok.