ultrakid / emqx_plugin_kafka Goto Github PK

你好，有升级emqx5.x的打算吗

我这边根据https://github.com/ULTRAKID/emqx 分支版本为main-v4.3构建成功并启动，开始发送消息到emqx时执行emqx_plugin_kafka插件报错。报错信息如下：
[error] [email protected]:49010 [Hooks] Failed to execute {emqx_plugin_kafka,on_message_publish,[[{broker,[{host,"172.16.3.168"},{port,"9092"},{partitionstrategy,"strict_round_robin"},{partitionworkers,1},{payloadtopic,"device_topic"}]}]]}: {exit,{{undef,[{pg2,create,[<<"ekaf.device_topic">>],[]},{ekaf_server,init,1,[{file,"ekaf_server.erl"},{line,69}]},{gen_fsm,init_it,6,[{file,"gen_fsm.erl"},{line,333}]},{proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,226}]}]},{gen_fsm,sync_send_event,[<0.2330.0>,prepare]}},[{gen_fsm,sync_send_event,2,[{file,"gen_fsm.erl"},{line,229}]},{ekaf_lib,prepare,2,[{file,"ekaf_lib.erl"},{line,55}]},{ekaf_lib,common_async,3,[{file,"ekaf_lib.erl"},{line,98}]},{emqx_plugin_kafka,on_message_publish,2,[{file,"emqx_plugin_kafka.erl"},{line,183}]},{emqx_hooks,safe_execute,2,[{file,"emqx_hooks.erl"},{line,191}]},{emqx_hooks,do_run_fold,3,[{file,"emqx_hooks.erl"},{line,170}]},{emqx_broker,publish,1,[{file,"emqx_broker.erl"},{line,204}]},{emqx_channel,do_publish,3,[{file,"emqx_channel.erl"},{line,578}]},{emqx_connection,with_channel,3,[{file,"emqx_connection.erl"},{line,664}]},{emqx_connection,process_msg,2,[{file,"emqx_connection.erl"},{line,364}]},{emqx_connection,process_msg,2,[{file,"emqx_connection.erl"},{line,370}]},{emqx_connection,handle_recv,3,[{file,"emqx_connection.erl"},{line,328}]},{proc_lib,wake_up,3,[{file,"proc_lib.erl"},{line,236}]}]}

配置信息为：

###kafka.port = 9092
kafka.port = 9092

The kafka loadbalancer node partition strategy.

##--------------------------------------------------------------------

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 = 172.16.3.168

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 = 1

payload topic.

Value: string

kafka.payloadtopic = device_topic

请问下我这边哪里出了问题，请指正。谢谢！

使用Releases中的4.3.12.5版本（ekaf）时，直接修改plugin下的配置文件即可。能正常使用。
但是4.3.12.6和7版本中，未找到该配置文件，emqx dashboard中也无法启动该插件。报错：parse_config_file_failed
docker logs：

2023-02-21T06:32:10.749386+00:00 [error] supervisor: {local,brod_sup}, errorContext: child_terminated, reason: [{{"localhost",9092},{econnrefused,[{kpro_connection,connect,4,[{file,"kpro_connection.erl"},{line,208}]},{kpro_connection,init,4,[{file,"kpro_connection.erl"},{line,170}]},{proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,226}]}]}}], offender: [{pid,<0.3742.0>},{id,emqx_repost_worker},{mfargs,{brod_client,start_link,[[{"localhost",9092}],emqx_repost_worker,[{reconnect_cool_down_seconds,10},{query_api_versions,true}]]}},{restart_type,{permanent,10}},{shutdown,5000},{child_type,worker}]

/emqx/_build/emqx/rel/emqx/etc/plugins

    [
      { emqx_plugin_kafka,[
          { kafka_address_list, [ { "172.28.57.6", 9093 } ] },
      { kafka_config, [
          { reconnect_cool_down_seconds, 10 },
      { query_api_versions, true },
      ]},
      { topic,<<"emqx-topic">> }
      ]}
    ]