From c3bc3cc51464fb53fd4d3e91317665f3997d7a95 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Thu, 16 May 2024 13:30:38 +0200
Subject: [PATCH 01/27] chore: ctl command run may return 'true'

---
 dev | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/dev b/dev
index 06249c843..a849f549e 100755
--- a/dev
+++ b/dev
@@ -506,6 +506,8 @@ ctl() {
         case rpc:call('$EMQX_NODE_NAME', emqx_ctl, run_command, [[$args]]) of
           ok ->
             init:stop(0);
+          true ->
+            init:stop(0);
           Error ->
             io:format(\"~p~n\", [Error]),
             init:stop(1)

From 331d44a78a7317e6ddd137e638590648412ceb77 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Sat, 18 May 2024 18:44:10 +0200
Subject: [PATCH 02/27] chore: improve Kafka producer health-check error logs

The Kafka producer lib wolff returns detailed error logs
about which host:port had issue, EMQX should log them to help
troubleshooting
---
 .../src/emqx_bridge_kafka_impl_producer.erl   | 69 ++++++++++++-------
 .../test/emqx_license_http_api_SUITE.erl      |  2 +-
 2 files changed, 46 insertions(+), 25 deletions(-)

diff --git a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
index 18ccd63aa..fa7bc67ac 100644
--- a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
+++ b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
@@ -572,33 +572,54 @@ check_client_connectivity(ClientId) ->
             {error, {find_client, Reason}}
     end.
 
+is_alive(Pid) ->
+    is_pid(Pid) andalso erlang:is_process_alive(Pid).
+
+error_summary(Map, [Error]) ->
+    Map#{error => Error};
+error_summary(Map, [Error | More]) ->
+    Map#{first_error => Error, total_errors => length(More) + 1}.
+
 check_if_healthy_leaders(ClientId, ClientPid, KafkaTopic, MaxPartitions) when is_pid(ClientPid) ->
-    Leaders =
-        case wolff_client:get_leader_connections(ClientPid, KafkaTopic, MaxPartitions) of
-            {ok, LeadersToCheck} ->
-                %% Kafka is considered healthy as long as any of the partition leader is reachable.
-                lists:filtermap(
-                    fun({_Partition, Pid}) ->
-                        case is_pid(Pid) andalso erlang:is_process_alive(Pid) of
-                            true -> {true, Pid};
-                            _ -> false
-                        end
-                    end,
-                    LeadersToCheck
-                );
-            {error, _} ->
-                []
-        end,
-    case Leaders of
-        [] ->
+    case wolff_client:get_leader_connections(ClientPid, KafkaTopic, MaxPartitions) of
+        {ok, Leaders} ->
+            %% Kafka is considered healthy as long as any of the partition leader is reachable.
+            case lists:partition(fun({_Partition, Pid}) -> is_alive(Pid) end, Leaders) of
+                {[], Errors} ->
+                    throw(
+                        error_summary(
+                            #{
+                                cause => "no_connected_partition_leader",
+                                kafka_client => ClientId,
+                                kafka_topic => KafkaTopic
+                            },
+                            Errors
+                        )
+                    );
+                {_, []} ->
+                    ok;
+                {_, Errors} ->
+                    ?SLOG(
+                        warning,
+                        "not_all_kafka_partitions_connected",
+                        error_summary(
+                            #{
+                                kafka_client => ClientId,
+                                kafka_topic => KafkaTopic
+                            },
+                            Errors
+                        )
+                    ),
+                    ok
+            end;
+        {error, Reason} ->
+            %% If failed to fetch metadata, wolff_client logs a warning level message
+            %% which includes the reason for each seed host
             throw(#{
-                error => no_connected_partition_leader,
+                cause => Reason,
                 kafka_client => ClientId,
-                kafka_topic => KafkaTopic,
-                partitions_limit => MaxPartitions
-            });
-        _ ->
-            ok
+                kafka_topic => KafkaTopic
+            })
     end.
 
 check_topic_status(ClientId, WolffClientPid, KafkaTopic) ->
diff --git a/apps/emqx_license/test/emqx_license_http_api_SUITE.erl b/apps/emqx_license/test/emqx_license_http_api_SUITE.erl
index 380930527..be0b31dcf 100644
--- a/apps/emqx_license/test/emqx_license_http_api_SUITE.erl
+++ b/apps/emqx_license/test/emqx_license_http_api_SUITE.erl
@@ -245,7 +245,7 @@ t_license_setting_bc(_Config) ->
     ?assertMatch(#{<<"max_connections">> := 25}, request_dump()),
     %% get
     GetRes = request(get, uri(["license", "setting"]), []),
-    %% aslo check that the settings return correctly
+    %% also check that the settings return correctly
     validate_setting(GetRes, <<"75%">>, <<"80%">>, 25),
     %% update
     Low = <<"50%">>,

From 7696f78d07d056e108595d352ad774d26401ee34 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Tue, 21 May 2024 09:27:12 +0200
Subject: [PATCH 03/27] docs: add changelog for pr 13070

---
 changes/ee/fix-13070.en.md | 5 +++++
 1 file changed, 5 insertions(+)
 create mode 100644 changes/ee/fix-13070.en.md

diff --git a/changes/ee/fix-13070.en.md b/changes/ee/fix-13070.en.md
new file mode 100644
index 000000000..e73cca3cf
--- /dev/null
+++ b/changes/ee/fix-13070.en.md
@@ -0,0 +1,5 @@
+Improve Kafka connector error logs.
+
+Previously, specific error details, such as unreachable advertised listeners, were not logged.
+Now, error details are captured in the logs to provide more diagnostic information.
+To manage log verbosity, only the first occurrence of an error is logged, accompanied by the total count of similar errors.

From f65168982b27ba41ab655980344968e79a940c8e Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 06:02:31 +0200
Subject: [PATCH 04/27] fix(oracle action trace): parameters should not be
 rendered as IO Data

This forces the parameters to the database statement to be rendered as
a JSON array in JSON traces instead of being rendered as a string when
the parameters are interpreted as IO data.

Fixes:
https://emqx.atlassian.net/browse/EMQX-12433
---
 apps/emqx/src/emqx_logger_jsonfmt.erl | 32 +++++++++++++++++++++++++++
 apps/emqx_oracle/src/emqx_oracle.erl  | 12 +++++++++-
 2 files changed, 43 insertions(+), 1 deletion(-)

diff --git a/apps/emqx/src/emqx_logger_jsonfmt.erl b/apps/emqx/src/emqx_logger_jsonfmt.erl
index 45697c1df..33763418c 100644
--- a/apps/emqx/src/emqx_logger_jsonfmt.erl
+++ b/apps/emqx/src/emqx_logger_jsonfmt.erl
@@ -270,6 +270,8 @@ json(L, Config) when is_list(L) ->
     end;
 json(Map, Config) when is_map(Map) ->
     best_effort_json_obj(Map, Config);
+json({'$array$', List}, Config) when is_list(List) ->
+    [json(I, Config) || I <- List];
 json(Term, Config) ->
     do_format_msg("~p", [Term], Config).
 
@@ -448,6 +450,36 @@ best_effort_json_test() ->
         <<"[\n  {\n    \"key\" : [\n      \n    ]\n  }\n]">>,
         best_effort_json([#{key => []}])
     ),
+    %% List is IO Data
+    ?assertMatch(
+        #{<<"what">> => <<"hej\n">>},
+        emqx_utils_json:decode(emqx_logger_jsonfmt:best_effort_json(#{what => [<<"hej">>, 10]}))
+    ),
+    %% Force list to be interpreted as an array
+    ?assertMatch(
+        #{<<"what">> => [<<"hej">>, 10]},
+        emqx_utils_json:decode(
+            emqx_logger_jsonfmt:best_effort_json(#{what => {'$array$', [<<"hej">>, 10]}})
+        )
+    ),
+    %% IO Data inside an array
+    ?assertMatch(
+        #{<<"what">> => [<<"hej">>, 10, <<"hej\n">>]},
+        emqx_utils_json:decode(
+            emqx_logger_jsonfmt:best_effort_json(#{
+                what => {'$array$', [<<"hej">>, 10, [<<"hej">>, 10]]}
+            })
+        )
+    ),
+    %% Array inside an array
+    ?assertMatch(
+        #{<<"what">> => [<<"hej">>, 10, [<<"hej">>, 10]]},
+        emqx_utils_json:decode(
+            emqx_logger_jsonfmt:best_effort_json(#{
+                what => {'$array$', [<<"hej">>, 10, {'$array$', [<<"hej">>, 10]}]}
+            })
+        )
+    ),
     ok.
 
 config() ->
diff --git a/apps/emqx_oracle/src/emqx_oracle.erl b/apps/emqx_oracle/src/emqx_oracle.erl
index e90665cc4..5b25e049a 100644
--- a/apps/emqx_oracle/src/emqx_oracle.erl
+++ b/apps/emqx_oracle/src/emqx_oracle.erl
@@ -8,6 +8,7 @@
 
 -include_lib("emqx_resource/include/emqx_resource.hrl").
 -include_lib("emqx/include/logger.hrl").
+-include_lib("emqx/include/emqx_trace.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 -define(UNHEALTHY_TARGET_MSG,
@@ -288,7 +289,7 @@ on_sql_query(InstId, ChannelID, PoolName, Type, ApplyMode, NameOrSQL, Data) ->
         type => Type,
         apply_mode => ApplyMode,
         name_or_sql => NameOrSQL,
-        data => Data
+        data => #emqx_trace_format_func_data{function = fun trace_format_data/1, data = Data}
     }),
     case ecpool:pick_and_do(PoolName, {?MODULE, Type, [NameOrSQL, Data]}, ApplyMode) of
         {error, Reason} = Result ->
@@ -317,6 +318,15 @@ on_sql_query(InstId, ChannelID, PoolName, Type, ApplyMode, NameOrSQL, Data) ->
             Result
     end.
 
+trace_format_data(Data0) ->
+    %% In batch request, we get a two level list
+    {'$array$', lists:map(fun insert_array_marker_if_list/1, Data0)}.
+
+insert_array_marker_if_list(List) when is_list(List) ->
+    {'$array$', List};
+insert_array_marker_if_list(Item) ->
+    Item.
+
 on_get_status(_InstId, #{pool_name := Pool} = _State) ->
     case emqx_resource_pool:health_check_workers(Pool, fun ?MODULE:do_get_status/1) of
         true ->

From e55bda6fe6ad884a771e9e6282ab776c56e6044f Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 10:56:59 +0200
Subject: [PATCH 05/27] chore: bump app versions

Bump app version as required to pass the check which is done by the
following script:

./scripts/apps-version-check.sh
---
 apps/emqx_oracle/src/emqx_oracle.app.src | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/emqx_oracle/src/emqx_oracle.app.src b/apps/emqx_oracle/src/emqx_oracle.app.src
index ce05bdfae..09aa4e589 100644
--- a/apps/emqx_oracle/src/emqx_oracle.app.src
+++ b/apps/emqx_oracle/src/emqx_oracle.app.src
@@ -1,6 +1,6 @@
 {application, emqx_oracle, [
     {description, "EMQX Enterprise Oracle Database Connector"},
-    {vsn, "0.2.0"},
+    {vsn, "0.2.1"},
     {registered, []},
     {applications, [
         kernel,

From e3cb63f59a41b17c17197ca3113743541a373f9f Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 11:08:07 +0200
Subject: [PATCH 06/27] docs: add change log entry

---
 changes/ee/fix-13136.en.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changes/ee/fix-13136.en.md

diff --git a/changes/ee/fix-13136.en.md b/changes/ee/fix-13136.en.md
new file mode 100644
index 000000000..8593c6fdb
--- /dev/null
+++ b/changes/ee/fix-13136.en.md
@@ -0,0 +1 @@
+The template-rendered traces for Oracle actions have been enhanced for better readability.

From 84bc6ba28cf9c37f4820469f979fc3b1f4d1d487 Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 10:43:32 +0200
Subject: [PATCH 07/27] fix: trace formatter should not crash when given a
 non-binary payload

Fixes:
https://emqx.atlassian.net/browse/EMQX-12474
---
 .../src/emqx_trace/emqx_trace_formatter.erl   | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 deletions(-)

diff --git a/apps/emqx/src/emqx_trace/emqx_trace_formatter.erl b/apps/emqx/src/emqx_trace/emqx_trace_formatter.erl
index 34469f835..e540ae82a 100644
--- a/apps/emqx/src/emqx_trace/emqx_trace_formatter.erl
+++ b/apps/emqx/src/emqx_trace/emqx_trace_formatter.erl
@@ -89,8 +89,17 @@ weight({packet, _}) -> {0, packet};
 weight({payload, _}) -> {2, payload};
 weight({K, _}) -> {1, K}.
 
-format_packet(undefined, _) -> "";
-format_packet(Packet, Encode) -> emqx_packet:format(Packet, Encode).
+format_packet(undefined, _) ->
+    "";
+format_packet(Packet, Encode) ->
+    try
+        emqx_packet:format(Packet, Encode)
+    catch
+        _:_ ->
+            %% We don't want to crash if there is a field named packet with
+            %% some other type of value
+            Packet
+    end.
 
 format_payload(undefined, _) ->
     "";
@@ -100,7 +109,11 @@ format_payload(Payload, text) when ?MAX_PAYLOAD_FORMAT_LIMIT(Payload) ->
     unicode:characters_to_list(Payload);
 format_payload(Payload, hex) when ?MAX_PAYLOAD_FORMAT_LIMIT(Payload) -> binary:encode_hex(Payload);
 format_payload(<<Part:?TRUNCATED_PAYLOAD_SIZE/binary, _/binary>> = Payload, Type) ->
-    emqx_packet:format_truncated_payload(Part, byte_size(Payload), Type).
+    emqx_packet:format_truncated_payload(Part, byte_size(Payload), Type);
+format_payload(Payload, _) ->
+    %% We don't want to crash if there is a field named payload with some other
+    %% type of value
+    Payload.
 
 to_iolist(Atom) when is_atom(Atom) -> atom_to_list(Atom);
 to_iolist(Int) when is_integer(Int) -> integer_to_list(Int);

From 2eb5c24068e617729c33381b9574c981a223b98f Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 10:54:04 +0200
Subject: [PATCH 08/27] docs: add change log entry

---
 changes/ce/fix-13140.en.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changes/ce/fix-13140.en.md

diff --git a/changes/ce/fix-13140.en.md b/changes/ce/fix-13140.en.md
new file mode 100644
index 000000000..921da9052
--- /dev/null
+++ b/changes/ce/fix-13140.en.md
@@ -0,0 +1 @@
+The issue causing text traces for the republish action to crash and not display correctly has been resolved.

From 5e3de1a4e3f204a45e4df7f4355307796a51087d Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 17:15:32 +0200
Subject: [PATCH 09/27] fix(emqx_logger_jsonfmt:best_effort_json_test): test
 cases

---
 apps/emqx/src/emqx_logger_jsonfmt.erl | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/apps/emqx/src/emqx_logger_jsonfmt.erl b/apps/emqx/src/emqx_logger_jsonfmt.erl
index 33763418c..d8dcc77dc 100644
--- a/apps/emqx/src/emqx_logger_jsonfmt.erl
+++ b/apps/emqx/src/emqx_logger_jsonfmt.erl
@@ -452,19 +452,19 @@ best_effort_json_test() ->
     ),
     %% List is IO Data
     ?assertMatch(
-        #{<<"what">> => <<"hej\n">>},
+        #{<<"what">> := <<"hej\n">>},
         emqx_utils_json:decode(emqx_logger_jsonfmt:best_effort_json(#{what => [<<"hej">>, 10]}))
     ),
     %% Force list to be interpreted as an array
     ?assertMatch(
-        #{<<"what">> => [<<"hej">>, 10]},
+        #{<<"what">> := [<<"hej">>, 10]},
         emqx_utils_json:decode(
             emqx_logger_jsonfmt:best_effort_json(#{what => {'$array$', [<<"hej">>, 10]}})
         )
     ),
     %% IO Data inside an array
     ?assertMatch(
-        #{<<"what">> => [<<"hej">>, 10, <<"hej\n">>]},
+        #{<<"what">> := [<<"hej">>, 10, <<"hej\n">>]},
         emqx_utils_json:decode(
             emqx_logger_jsonfmt:best_effort_json(#{
                 what => {'$array$', [<<"hej">>, 10, [<<"hej">>, 10]]}
@@ -473,7 +473,7 @@ best_effort_json_test() ->
     ),
     %% Array inside an array
     ?assertMatch(
-        #{<<"what">> => [<<"hej">>, 10, [<<"hej">>, 10]]},
+        #{<<"what">> := [<<"hej">>, 10, [<<"hej">>, 10]]},
         emqx_utils_json:decode(
             emqx_logger_jsonfmt:best_effort_json(#{
                 what => {'$array$', [<<"hej">>, 10, {'$array$', [<<"hej">>, 10]}]}

From cbdde7165ece1c2d11acd9a291ea310b3117a0de Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Mon, 27 May 2024 15:36:20 +0200
Subject: [PATCH 10/27] fix(redis trace): add separators in redis batch action
 trace

Logger will transform data that looks like IO data into a string which
made redis batch traces look like the spaces had been removed
from the strings. To prevent this, we render the batched commands into
binary string separated by spaces and semicolon. The components of a
single command is separated by spaces and the commands in a batch are
separated by semicolons.

Fixes:
https://emqx.atlassian.net/browse/EMQX-12428
---
 .../src/emqx_bridge_redis_connector.erl            | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl b/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
index 535d6e13c..c98863f33 100644
--- a/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
+++ b/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
@@ -6,6 +6,7 @@
 -include_lib("emqx/include/logger.hrl").
 -include_lib("emqx_resource/include/emqx_resource.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx/include/emqx_trace.hrl").
 
 -behaviour(emqx_resource).
 
@@ -143,7 +144,13 @@ on_batch_query(
             [{ChannelID, _} | _] = BatchData,
             emqx_trace:rendered_action_template(
                 ChannelID,
-                #{commands => Cmds, batch => ture}
+                #{
+                    commands => #emqx_trace_format_func_data{
+                        function = fun trace_format_commands/1,
+                        data = Cmds
+                    },
+                    batch => true
+                }
             ),
             Result = query(InstId, {cmds, Cmds}, RedisConnSt),
             ?tp(
@@ -162,6 +169,11 @@ on_batch_query(
             Error
     end.
 
+trace_format_commands(Commands0) ->
+    Commands1 = [[unicode:characters_to_list(S) || S <- C] || C <- Commands0],
+    Commands2 = [lists:join(" ", C) || C <- Commands1],
+    unicode:characters_to_binary(lists:join("; ", Commands2)).
+
 on_format_query_result({ok, Msg}) ->
     #{result => ok, message => Msg};
 on_format_query_result(Res) ->

From 4099e22a7732c367cfa408842876c3ad2f0ceff5 Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Mon, 27 May 2024 15:47:48 +0200
Subject: [PATCH 11/27] docs: add changelog entry

---
 changes/ee/fix-13130.en.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changes/ee/fix-13130.en.md

diff --git a/changes/ee/fix-13130.en.md b/changes/ee/fix-13130.en.md
new file mode 100644
index 000000000..7c111cacd
--- /dev/null
+++ b/changes/ee/fix-13130.en.md
@@ -0,0 +1 @@
+Traces for Redis action batch requests have got improved formatting. Spaces are now added between components of commands and semicolons are added between commands to make the trace message easier to read.

From 3f1fe413eef9fc788b46133daa8125db26911fae Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 06:17:03 +0200
Subject: [PATCH 12/27] refactor: simplify redis trace formatting

---
 apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl b/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
index c98863f33..e12155bb1 100644
--- a/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
+++ b/apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
@@ -170,9 +170,8 @@ on_batch_query(
     end.
 
 trace_format_commands(Commands0) ->
-    Commands1 = [[unicode:characters_to_list(S) || S <- C] || C <- Commands0],
-    Commands2 = [lists:join(" ", C) || C <- Commands1],
-    unicode:characters_to_binary(lists:join("; ", Commands2)).
+    Commands1 = [lists:join(" ", C) || C <- Commands0],
+    unicode:characters_to_binary(lists:join("; ", Commands1)).
 
 on_format_query_result({ok, Msg}) ->
     #{result => ok, message => Msg};

From 3151a3123e7c74c68030a4af7ec76d765c35239f Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 17:23:35 +0200
Subject: [PATCH 13/27] chore: bump version of emqx_redis_bridge app

---
 apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src b/apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src
index 57b4d90f1..acd1837ba 100644
--- a/apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src
+++ b/apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src
@@ -1,6 +1,6 @@
 {application, emqx_bridge_redis, [
     {description, "EMQX Enterprise Redis Bridge"},
-    {vsn, "0.1.6"},
+    {vsn, "0.1.7"},
     {registered, []},
     {applications, [
         kernel,

From 82800faadf3debe9db5f26493b660b875624cc25 Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 18:07:14 +0200
Subject: [PATCH 14/27] fix: make protobuf schema decode error reported to user
 less cryptic

Before this commit the user would see a cryptic warning log including
function_clause when decoding a message failed. This commit improves
this by removing function_caluse as well as some other cryptic words
from the message and adding a description describing what went wrong.

Fixes:
https://emqx.atlassian.net/browse/EMQX-12453
---
 .../src/emqx_schema_registry.app.src             |  2 +-
 .../src/emqx_schema_registry_serde.erl           | 16 +++++++++++++++-
 2 files changed, 16 insertions(+), 2 deletions(-)

diff --git a/apps/emqx_schema_registry/src/emqx_schema_registry.app.src b/apps/emqx_schema_registry/src/emqx_schema_registry.app.src
index f05295579..efd9f1162 100644
--- a/apps/emqx_schema_registry/src/emqx_schema_registry.app.src
+++ b/apps/emqx_schema_registry/src/emqx_schema_registry.app.src
@@ -1,6 +1,6 @@
 {application, emqx_schema_registry, [
     {description, "EMQX Schema Registry"},
-    {vsn, "0.3.0"},
+    {vsn, "0.3.1"},
     {registered, [emqx_schema_registry_sup]},
     {mod, {emqx_schema_registry_app, []}},
     {included_applications, [
diff --git a/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl b/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl
index 638147fd1..96c5816ba 100644
--- a/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl
+++ b/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl
@@ -64,7 +64,21 @@ handle_rule_function(sparkplug_encode, [Term | MoreArgs]) ->
         [?EMQX_SCHEMA_REGISTRY_SPARKPLUGB_SCHEMA_NAME, Term | MoreArgs]
     );
 handle_rule_function(schema_decode, [SchemaId, Data | MoreArgs]) ->
-    decode(SchemaId, Data, MoreArgs);
+    try
+        decode(SchemaId, Data, MoreArgs)
+    catch
+        error:{gpb_error, {decoding_failure, {_Data, _Schema, {error, function_clause, _Stack}}}} ->
+            throw(
+                {schema_decode_error, #{
+                    error_type => decoding_failure,
+                    schema_id => SchemaId,
+                    data => Data,
+                    more_args => MoreArgs,
+                    msg =>
+                        <<"The given data could not be decoded. Please check the input data and the schema.">>
+                }}
+            )
+    end;
 handle_rule_function(schema_decode, Args) ->
     error({args_count_error, {schema_decode, Args}});
 handle_rule_function(schema_encode, [SchemaId, Term | MoreArgs]) ->

From 9d7acc6a3289f34d449fe3b7e473ce13ae44c306 Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Tue, 28 May 2024 18:24:45 +0200
Subject: [PATCH 15/27] docs: add change log entry

---
 changes/ee/fix-13147.en.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changes/ee/fix-13147.en.md

diff --git a/changes/ee/fix-13147.en.md b/changes/ee/fix-13147.en.md
new file mode 100644
index 000000000..4f717892d
--- /dev/null
+++ b/changes/ee/fix-13147.en.md
@@ -0,0 +1 @@
+Error messages for decoding failures in the rule engine protobuf decode functions have been improved by adding a clear descriptions to indicate what went wrong when message decoding fails.

From 00d74826ec3566f134b70cb8be9cb1d14d089efd Mon Sep 17 00:00:00 2001
From: zhongwencool <zhongwencool@gmail.com>
Date: Fri, 17 May 2024 10:38:11 +0800
Subject: [PATCH 16/27] chore: improved InfluxDB connection error messaging

---
 .../src/emqx_bridge_influxdb_connector.erl                    | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl b/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl
index 88065b7b3..852f78485 100644
--- a/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl
+++ b/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl
@@ -396,7 +396,7 @@ do_start_client(InstId, ClientConfig, Config) ->
                             }),
                             %% no leak
                             _ = influxdb:stop_client(Client),
-                            {error, influxdb_client_auth_error}
+                            {error, connect_ok_but_auth_failed}
                     end;
                 {false, Reason} ->
                     ?tp(influxdb_connector_start_failed, #{
@@ -410,7 +410,7 @@ do_start_client(InstId, ClientConfig, Config) ->
                     }),
                     %% no leak
                     _ = influxdb:stop_client(Client),
-                    {error, influxdb_client_not_alive}
+                    {error, {connect_failed, Reason}}
             end;
         {error, {already_started, Client0}} ->
             ?tp(influxdb_connector_start_already_started, #{}),

From 7d9333a9d7ab753180beef6b7ebedaf4a43fc244 Mon Sep 17 00:00:00 2001
From: zhongwencool <zhongwencool@gmail.com>
Date: Tue, 28 May 2024 16:42:42 +0800
Subject: [PATCH 17/27] chore: bump influxdb app version

---
 apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src b/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src
index d65a0bf11..e211498ac 100644
--- a/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src
+++ b/apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src
@@ -1,6 +1,6 @@
 {application, emqx_bridge_influxdb, [
     {description, "EMQX Enterprise InfluxDB Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {applications, [
         kernel,

From 9b089a4e5aa44c3013743a325002c21d1a0e5bc0 Mon Sep 17 00:00:00 2001
From: Kjell Winblad <kjellwinblad@gmail.com>
Date: Wed, 29 May 2024 10:20:15 +0200
Subject: [PATCH 18/27] fix: use explain instead of msg as error explanation
 and add test

---
 .../src/emqx_schema_registry_serde.erl        |  2 +-
 .../test/emqx_schema_registry_SUITE.erl       | 20 ++++++++++++++++++-
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl b/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl
index 96c5816ba..e8da35449 100644
--- a/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl
+++ b/apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl
@@ -74,7 +74,7 @@ handle_rule_function(schema_decode, [SchemaId, Data | MoreArgs]) ->
                     schema_id => SchemaId,
                     data => Data,
                     more_args => MoreArgs,
-                    msg =>
+                    explain =>
                         <<"The given data could not be decoded. Please check the input data and the schema.">>
                 }}
             )
diff --git a/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl b/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl
index 22252b7c3..d9286f266 100644
--- a/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl
+++ b/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl
@@ -44,7 +44,8 @@ sparkplug_tests() ->
         t_sparkplug_decode,
         t_sparkplug_encode,
         t_sparkplug_decode_encode_with_message_name,
-        t_sparkplug_encode_float_to_uint64_key
+        t_sparkplug_encode_float_to_uint64_key,
+        t_decode_fail
     ].
 
 init_per_suite(Config) ->
@@ -532,6 +533,23 @@ t_encode(Config) ->
     end,
     ok.
 
+t_decode_fail(_Config) ->
+    SerdeName = my_serde,
+    SerdeType = protobuf,
+    ok = create_serde(SerdeType, SerdeName),
+    Payload = <<"ss">>,
+    ?assertThrow(
+        {schema_decode_error, #{
+            data := <<"ss">>,
+            error_type := decoding_failure,
+            explain := _,
+            more_args := [<<"Person">>],
+            schema_id := <<"my_serde">>
+        }},
+        emqx_rule_funcs:schema_decode(<<"my_serde">>, Payload, <<"Person">>)
+    ),
+    ok.
+
 t_decode(Config) ->
     SerdeType = ?config(serde_type, Config),
     SerdeName = my_serde,

From 1db9d54d1733593767856eb4aee7f6c9f3f529c8 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Mon, 27 May 2024 16:29:33 +0200
Subject: [PATCH 19/27] fix: handle unicode in data integration template
 strings

The bug was introduced in 5.7.0 (#12893)
---
 apps/emqx/src/emqx_schema.erl        | 26 ++++++++++++++------------
 apps/emqx/test/emqx_schema_tests.erl | 10 ++++++++++
 2 files changed, 24 insertions(+), 12 deletions(-)

diff --git a/apps/emqx/src/emqx_schema.erl b/apps/emqx/src/emqx_schema.erl
index 948849ccb..da8b885c6 100644
--- a/apps/emqx/src/emqx_schema.erl
+++ b/apps/emqx/src/emqx_schema.erl
@@ -2471,7 +2471,7 @@ converter_ciphers(Ciphers, _Opts) when is_binary(Ciphers) ->
 
 default_ciphers(Which) ->
     lists:map(
-        fun erlang:iolist_to_binary/1,
+        fun unicode:characters_to_binary/1,
         do_default_ciphers(Which)
     ).
 
@@ -2494,7 +2494,7 @@ bin_str_converter(I, _) when is_integer(I) ->
     integer_to_binary(I);
 bin_str_converter(X, _) ->
     try
-        iolist_to_binary(X)
+        unicode:characters_to_binary(X)
     catch
         _:_ ->
             throw("must_quote")
@@ -2649,7 +2649,7 @@ to_comma_separated_list(Str) ->
     {ok, string:tokens(Str, ", ")}.
 
 to_comma_separated_binary(Str) ->
-    {ok, lists:map(fun erlang:list_to_binary/1, string:tokens(Str, ", "))}.
+    {ok, lists:map(fun unicode:characters_to_binary/1, string:tokens(Str, ", "))}.
 
 to_comma_separated_atoms(Str) ->
     {ok, lists:map(fun to_atom/1, string:tokens(Str, ", "))}.
@@ -2658,7 +2658,7 @@ to_url(Str) ->
     case emqx_http_lib:uri_parse(Str) of
         {ok, URIMap} ->
             URIString = emqx_http_lib:normalize(URIMap),
-            {ok, iolist_to_binary(URIString)};
+            {ok, unicode:characters_to_binary(URIString)};
         Error ->
             Error
     end.
@@ -2666,13 +2666,13 @@ to_url(Str) ->
 to_json_binary(Str) ->
     case emqx_utils_json:safe_decode(Str) of
         {ok, _} ->
-            {ok, iolist_to_binary(Str)};
+            {ok, unicode:characters_to_binary(Str)};
         Error ->
             Error
     end.
 
 to_template(Str) ->
-    {ok, iolist_to_binary(Str)}.
+    {ok, unicode:characters_to_binary(Str, utf8)}.
 
 to_template_str(Str) ->
     {ok, unicode:characters_to_list(Str, utf8)}.
@@ -2768,7 +2768,7 @@ validate_keepalive_multiplier(_Multiplier) ->
     {error, #{reason => keepalive_multiplier_out_of_range, min => 1, max => 65535}}.
 
 validate_tcp_keepalive(Value) ->
-    case iolist_to_binary(Value) of
+    case unicode:characters_to_binary(Value) of
         <<"none">> ->
             ok;
         _ ->
@@ -2949,7 +2949,7 @@ convert_servers(undefined) ->
 convert_servers(Map) when is_map(Map) ->
     try
         List = convert_hocon_map_host_port(Map),
-        iolist_to_binary(string:join(List, ","))
+        unicode:characters_to_binary(string:join(List, ","))
     catch
         _:_ ->
             throw("bad_host_port")
@@ -2957,13 +2957,13 @@ convert_servers(Map) when is_map(Map) ->
 convert_servers([H | _] = Array) when is_binary(H) orelse is_list(H) ->
     %% if the old config was a string array
     %% we want to make sure it's converted to a comma-separated
-    iolist_to_binary([[I, ","] || I <- Array]);
+    unicode:characters_to_binary([[I, ","] || I <- Array]);
 convert_servers(Str) ->
     normalize_host_port_str(Str).
 
 %% remove spaces around comma (,)
 normalize_host_port_str(Str) ->
-    iolist_to_binary(re:replace(Str, "(\s)*,(\s)*", ",")).
+    unicode:characters_to_binary(re:replace(Str, "(\s)*,(\s)*", ",")).
 
 %% @doc Shared validation function for both 'server' and 'servers' string.
 %% NOTE: Validator is called after converter.
@@ -3442,8 +3442,10 @@ ensure_default_listener(Map, ListenerType) ->
     NewMap = Map#{<<"default">> => default_listener(ListenerType)},
     keep_default_tombstone(NewMap, #{}).
 
-cert_file(_File, client) -> undefined;
-cert_file(File, server) -> iolist_to_binary(filename:join(["${EMQX_ETC_DIR}", "certs", File])).
+cert_file(_File, client) ->
+    undefined;
+cert_file(File, server) ->
+    unicode:characters_to_binary(filename:join(["${EMQX_ETC_DIR}", "certs", File])).
 
 mqtt_converter(#{<<"keepalive_multiplier">> := Multi} = Mqtt, _Opts) ->
     case round(Multi * 100) =:= round(?DEFAULT_MULTIPLIER * 100) of
diff --git a/apps/emqx/test/emqx_schema_tests.erl b/apps/emqx/test/emqx_schema_tests.erl
index 2f3f2aa1d..b94815995 100644
--- a/apps/emqx/test/emqx_schema_tests.erl
+++ b/apps/emqx/test/emqx_schema_tests.erl
@@ -930,3 +930,13 @@ timeout_types_test_() ->
             typerefl:from_string(emqx_schema:timeout_duration_s(), <<"4294967001ms">>)
         )
     ].
+
+unicode_template_test() ->
+    Sc = #{
+        roots => [root],
+        fields => #{root => [{template, #{type => emqx_schema:template()}}]}
+    },
+    HoconText = <<"root = {template = \"中文\"}"/utf8>>,
+    {ok, Hocon} = hocon:binary(HoconText),
+    _ = hocon_tconf:check_plain(Sc, Hocon),
+    ok.

From 78a6100346ceb837bcab156d0f7bf456c00586e1 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Wed, 29 May 2024 21:56:22 +0200
Subject: [PATCH 20/27] chore: fix app vsn bumps

only bug fixes so far
---
 apps/emqx/src/emqx.app.src                       | 2 +-
 apps/emqx_auth/src/emqx_auth.app.src             | 2 +-
 apps/emqx_management/src/emqx_management.app.src | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/apps/emqx/src/emqx.app.src b/apps/emqx/src/emqx.app.src
index c391bc1c8..97769fe1f 100644
--- a/apps/emqx/src/emqx.app.src
+++ b/apps/emqx/src/emqx.app.src
@@ -2,7 +2,7 @@
 {application, emqx, [
     {id, "emqx"},
     {description, "EMQX Core"},
-    {vsn, "5.4.0"},
+    {vsn, "5.3.1"},
     {modules, []},
     {registered, []},
     {applications, [
diff --git a/apps/emqx_auth/src/emqx_auth.app.src b/apps/emqx_auth/src/emqx_auth.app.src
index 35f3ff8b0..db4b69ded 100644
--- a/apps/emqx_auth/src/emqx_auth.app.src
+++ b/apps/emqx_auth/src/emqx_auth.app.src
@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_auth, [
     {description, "EMQX Authentication and authorization"},
-    {vsn, "0.4.0"},
+    {vsn, "0.3.1"},
     {modules, []},
     {registered, [emqx_auth_sup]},
     {applications, [
diff --git a/apps/emqx_management/src/emqx_management.app.src b/apps/emqx_management/src/emqx_management.app.src
index f01f5c7b9..7c99b70cd 100644
--- a/apps/emqx_management/src/emqx_management.app.src
+++ b/apps/emqx_management/src/emqx_management.app.src
@@ -2,7 +2,7 @@
 {application, emqx_management, [
     {description, "EMQX Management API and CLI"},
     % strict semver, bump manually!
-    {vsn, "5.3.0"},
+    {vsn, "5.2.1"},
     {modules, []},
     {registered, [emqx_management_sup]},
     {applications, [

From 9fe29bf5f748252493d2825e97f0731fc6038265 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Wed, 29 May 2024 21:57:08 +0200
Subject: [PATCH 21/27] chore: bump kafka bridge app vsn

---
 apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src
index d7af76b8b..ea7bfaba8 100644
--- a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src
+++ b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src
@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge_kafka, [
     {description, "EMQX Enterprise Kafka Bridge"},
-    {vsn, "0.3.0"},
+    {vsn, "0.3.1"},
     {registered, [emqx_bridge_kafka_consumer_sup]},
     {applications, [
         kernel,

From 532812f6f6efafbcec81c0ceadc48f216d7a9028 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Wed, 29 May 2024 22:10:34 +0200
Subject: [PATCH 22/27] test: assert results

---
 apps/emqx/test/emqx_schema_tests.erl | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/apps/emqx/test/emqx_schema_tests.erl b/apps/emqx/test/emqx_schema_tests.erl
index b94815995..8f72c3d48 100644
--- a/apps/emqx/test/emqx_schema_tests.erl
+++ b/apps/emqx/test/emqx_schema_tests.erl
@@ -938,5 +938,7 @@ unicode_template_test() ->
     },
     HoconText = <<"root = {template = \"中文\"}"/utf8>>,
     {ok, Hocon} = hocon:binary(HoconText),
-    _ = hocon_tconf:check_plain(Sc, Hocon),
-    ok.
+    ?assertEqual(
+        #{<<"root">> => #{<<"template">> => <<"中文"/utf8>>}},
+        hocon_tconf:check_plain(Sc, Hocon)
+    ).

From a00d9f17b1b8cac8923b8663c7562e4d58736bdf Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Tue, 21 May 2024 09:04:18 +0200
Subject: [PATCH 23/27] chore: upgrade kafka producer lib wolff to 1.10.4

wolff-1.10.4 handles message_too_large error code from Kafka.
It tries to split the too-large batch into single-call batches
(for EMQX, one call batch is always one message), and retry.
In case a single-call batch is too large, EMQX should increment
a failure counter.
---
 apps/emqx_bridge_azure_event_hub/rebar.config               | 2 +-
 apps/emqx_bridge_confluent/rebar.config                     | 2 +-
 apps/emqx_bridge_kafka/rebar.config                         | 2 +-
 .../src/emqx_bridge_kafka_impl_producer.erl                 | 6 +++---
 changes/ee/fix-13079.en.md                                  | 6 ++++++
 mix.exs                                                     | 2 +-
 6 files changed, 13 insertions(+), 7 deletions(-)
 create mode 100644 changes/ee/fix-13079.en.md

diff --git a/apps/emqx_bridge_azure_event_hub/rebar.config b/apps/emqx_bridge_azure_event_hub/rebar.config
index 269239620..8cd5ee427 100644
--- a/apps/emqx_bridge_azure_event_hub/rebar.config
+++ b/apps/emqx_bridge_azure_event_hub/rebar.config
@@ -2,7 +2,7 @@
 
 {erl_opts, [debug_info]}.
 {deps, [
-    {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.2"}}},
+    {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.4"}}},
     {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}},
     {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}},
     {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}},
diff --git a/apps/emqx_bridge_confluent/rebar.config b/apps/emqx_bridge_confluent/rebar.config
index 0519e39c9..a969ac83b 100644
--- a/apps/emqx_bridge_confluent/rebar.config
+++ b/apps/emqx_bridge_confluent/rebar.config
@@ -2,7 +2,7 @@
 
 {erl_opts, [debug_info]}.
 {deps, [
-    {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.2"}}},
+    {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.4"}}},
     {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}},
     {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}},
     {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}},
diff --git a/apps/emqx_bridge_kafka/rebar.config b/apps/emqx_bridge_kafka/rebar.config
index 7c98bf571..fd7f03da8 100644
--- a/apps/emqx_bridge_kafka/rebar.config
+++ b/apps/emqx_bridge_kafka/rebar.config
@@ -2,7 +2,7 @@
 
 {erl_opts, [debug_info]}.
 {deps, [
-    {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.2"}}},
+    {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.4"}}},
     {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}},
     {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}},
     {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}},
diff --git a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
index fa7bc67ac..53f2603d4 100644
--- a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
+++ b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
@@ -211,7 +211,7 @@ ensure_client(ClientId, Hosts, ClientConfig) ->
     case wolff_client_sup:find_client(ClientId) of
         {ok, _Pid} ->
             ok;
-        {error, no_such_client} ->
+        {error, #{reason := no_such_client}} ->
             case wolff:ensure_supervised_client(ClientId, Hosts, ClientConfig) of
                 {ok, _} ->
                     ?SLOG(info, #{
@@ -543,13 +543,13 @@ check_topic_and_leader_connections(ClientId, KafkaTopic, MaxPartitions) ->
         {ok, Pid} ->
             ok = check_topic_status(ClientId, Pid, KafkaTopic),
             ok = check_if_healthy_leaders(ClientId, Pid, KafkaTopic, MaxPartitions);
-        {error, no_such_client} ->
+        {error, #{reason := no_such_client}} ->
             throw(#{
                 reason => cannot_find_kafka_client,
                 kafka_client => ClientId,
                 kafka_topic => KafkaTopic
             });
-        {error, client_supervisor_not_initialized} ->
+        {error, #{reason := client_supervisor_not_initialized}} ->
             throw(#{
                 reason => restarting,
                 kafka_client => ClientId,
diff --git a/changes/ee/fix-13079.en.md b/changes/ee/fix-13079.en.md
new file mode 100644
index 000000000..c469bdf6e
--- /dev/null
+++ b/changes/ee/fix-13079.en.md
@@ -0,0 +1,6 @@
+Improve Kafka producer error handling for `message_too_large`.
+
+Prior to this change, Kafka producers would retry sending oversized batches (`message_too_large` error) in hopes of a server side configuration fix (`max.message.bytes`).
+
+Now, oversized messages are automatically split into single-message batches for retry.
+If a message still exceeds size limits, it will be dropped to maintain data flow.
diff --git a/mix.exs b/mix.exs
index f3e91096e..85aa88d62 100644
--- a/mix.exs
+++ b/mix.exs
@@ -210,7 +210,7 @@ defmodule EMQXUmbrella.MixProject do
       {:hstreamdb_erl,
        github: "hstreamdb/hstreamdb_erl", tag: "0.5.18+v0.18.1+ezstd-v1.0.5-emqx1"},
       {:influxdb, github: "emqx/influxdb-client-erl", tag: "1.1.13", override: true},
-      {:wolff, github: "kafka4beam/wolff", tag: "1.10.2"},
+      {:wolff, github: "kafka4beam/wolff", tag: "1.10.4"},
       {:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.5", override: true},
       {:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.1"},
       {:brod, github: "kafka4beam/brod", tag: "3.16.8"},

From 38f49de6d96239793f5207fc26f0826f6f8c1707 Mon Sep 17 00:00:00 2001
From: Ivan Dyachkov <dev@dyachkov.org>
Date: Wed, 29 May 2024 15:28:55 +0200
Subject: [PATCH 24/27] fix(deb): fix apt purge when emqx service was still
 running

---
 deploy/packages/deb/debian/postrm |  5 +++--
 scripts/pkg-tests.sh              | 15 +++++++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/deploy/packages/deb/debian/postrm b/deploy/packages/deb/debian/postrm
index 5df55135a..9ee1dca87 100755
--- a/deploy/packages/deb/debian/postrm
+++ b/deploy/packages/deb/debian/postrm
@@ -21,6 +21,8 @@ set -e
 
 case "$1" in
     purge)
+        # force kill all processes owned by emqx, if any
+        pkill -9 -u emqx || true
         rm -f /etc/default/emqx
 
         if [ -d /var/lib/emqx ]; then
@@ -38,9 +40,8 @@ case "$1" in
         if [ -e /etc/init.d/emqx ]; then
                 rm  /etc/init.d/emqx
         fi
-        # Remove User & Group, killing any process owned by them
+        # Remove User & Group
         if getent passwd emqx >/dev/null; then
-                pkill -u emqx || true
                 deluser --quiet --system emqx
         fi
         if getent group emqx >/dev/null; then
diff --git a/scripts/pkg-tests.sh b/scripts/pkg-tests.sh
index a88020dd5..dfb5d098f 100755
--- a/scripts/pkg-tests.sh
+++ b/scripts/pkg-tests.sh
@@ -131,6 +131,21 @@ emqx_test(){
                 exit 1
             fi
 
+            echo "try to install again and purge while the service is running"
+            dpkg -i "${PACKAGE_PATH}/${packagename}"
+            if [ "$(dpkg -l | grep ${EMQX_NAME} | awk '{print $1}')" != "ii" ]
+            then
+                echo "package install error"
+                exit 1
+            fi
+            if ! /usr/bin/emqx start
+            then
+                echo "ERROR: failed_to_start_emqx"
+                cat /var/log/emqx/erlang.log.1 || true
+                cat /var/log/emqx/emqx.log.1 || true
+                exit 1
+            fi
+            /usr/bin/emqx ping
             dpkg -P "${EMQX_NAME}"
             if dpkg -l |grep -q emqx
             then

From e54395f5018a9577860fcf9f499201e95728c307 Mon Sep 17 00:00:00 2001
From: Ivan Dyachkov <dev@dyachkov.org>
Date: Wed, 29 May 2024 13:28:21 +0200
Subject: [PATCH 25/27] ci: always trigger package reindex by emqx.io on
 releases

---
 .github/workflows/release.yaml | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
index f40db0039..b3c5e4613 100644
--- a/.github/workflows/release.yaml
+++ b/.github/workflows/release.yaml
@@ -8,7 +8,7 @@ on:
       tag:
         type: string
         required: true
-      publish_release_artefacts:
+      publish_release_artifacts:
         type: boolean
         required: true
         default: false
@@ -75,7 +75,7 @@ jobs:
           tag_name: "${{ env.ref_name }}"
           skip_existing: true
       - name: update to emqx.io
-        if: startsWith(env.ref_name, 'v') && ((github.event_name == 'release' && !github.event.release.prerelease) || inputs.publish_release_artefacts)
+        if: github.event_name == 'release' || inputs.publish_release_artifacts
         run: |
           set -eux
           curl -w %{http_code} \
@@ -86,7 +86,7 @@ jobs:
                -d "{\"repo\":\"emqx/emqx\", \"tag\": \"${{ env.ref_name }}\" }" \
                ${{ secrets.EMQX_IO_RELEASE_API }}
       - name: Push to packagecloud.io
-        if: (github.event_name == 'release' && !github.event.release.prerelease) || inputs.publish_release_artefacts
+        if: (github.event_name == 'release' && !github.event.release.prerelease) || inputs.publish_release_artifacts
         env:
           PROFILE: ${{ steps.profile.outputs.profile }}
           VERSION: ${{ steps.profile.outputs.version }}

From 830926044c32ff6b1adfbd0a3b182232700b61e1 Mon Sep 17 00:00:00 2001
From: Ivan Dyachkov <dev@dyachkov.org>
Date: Thu, 30 May 2024 17:02:03 +0200
Subject: [PATCH 26/27] chore: bump erlang-rocksdb to 1.8.0-emqx-5

- fix macos build https://github.com/google/snappy/issues/183
- pre-built binaries for latest platforms
---
 mix.exs      | 2 +-
 rebar.config | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/mix.exs b/mix.exs
index f3e91096e..7b17931fb 100644
--- a/mix.exs
+++ b/mix.exs
@@ -54,7 +54,7 @@ defmodule EMQXUmbrella.MixProject do
       {:jiffy, github: "emqx/jiffy", tag: "1.0.6", override: true},
       {:cowboy, github: "emqx/cowboy", tag: "2.9.2", override: true},
       {:esockd, github: "emqx/esockd", tag: "5.11.2", override: true},
-      {:rocksdb, github: "emqx/erlang-rocksdb", tag: "1.8.0-emqx-2", override: true},
+      {:rocksdb, github: "emqx/erlang-rocksdb", tag: "1.8.0-emqx-5", override: true},
       {:ekka, github: "emqx/ekka", tag: "0.19.3", override: true},
       {:gen_rpc, github: "emqx/gen_rpc", tag: "3.3.1", override: true},
       {:grpc, github: "emqx/grpc-erl", tag: "0.6.12", override: true},
diff --git a/rebar.config b/rebar.config
index 98f8e2177..b260b366e 100644
--- a/rebar.config
+++ b/rebar.config
@@ -82,7 +82,7 @@
     {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.6"}}},
     {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}},
     {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.11.2"}}},
-    {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb", {tag, "1.8.0-emqx-2"}}},
+    {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb", {tag, "1.8.0-emqx-5"}}},
     {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.19.3"}}},
     {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.1"}}},
     {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.12"}}},

From 2d5d96ae331b896ab4aae4418f2583ead3f5e547 Mon Sep 17 00:00:00 2001
From: Ivan Dyachkov <dev@dyachkov.org>
Date: Thu, 30 May 2024 18:27:29 +0200
Subject: [PATCH 27/27] ci(ui-tests): retry getting element by xpath

---
 scripts/ui-tests/dashboard_test.py | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/scripts/ui-tests/dashboard_test.py b/scripts/ui-tests/dashboard_test.py
index 602d944fd..85d40cec8 100644
--- a/scripts/ui-tests/dashboard_test.py
+++ b/scripts/ui-tests/dashboard_test.py
@@ -100,7 +100,15 @@ def test_docs_link(driver, login, dashboard_url):
     driver.get(dest_url)
     ensure_current_url(driver, dest_url)
     xpath_link_help = "//div[@id='app']//div[@class='nav-header']//a[contains(@class, 'link-help')]"
-    link_help = driver.find_element(By.XPATH, xpath_link_help)
+    # retry up to 5 times
+    for _ in range(5):
+        try:
+            link_help = driver.find_element(By.XPATH, xpath_link_help)
+            break
+        except NoSuchElementException:
+            time.sleep(1)
+    else:
+        raise AssertionError("Cannot find the help link")
     driver.execute_script("arguments[0].click();", link_help)
 
     prefix, emqx_version = fetch_version(dashboard_url)