fix(message transformation): improve error messages when payload to decode is not binary

Fixes https://emqx.atlassian.net/browse/EMQX-12931

apps/emqx_message_transformation/src/emqx_message_transformation.app.src

@@ -1,6 +1,6 @@
 {application, emqx_message_transformation, [
     {description, "EMQX Message Transformation"},
-    {vsn, "0.1.2"},
+    {vsn, "0.1.3"},
     {registered, [emqx_message_transformation_sup, emqx_message_transformation_registry]},
     {mod, {emqx_message_transformation_app, []}},
     {applications, [

apps/emqx_message_transformation/src/emqx_message_transformation.erl

@@ -446,7 +446,7 @@ take_from_context(Context, Message) ->
 
 decode(Payload, #{type := none}, _Transformation) ->
     {ok, Payload};
-decode(Payload, #{type := json}, Transformation) ->
+decode(Payload, #{type := json}, Transformation) when is_binary(Payload) ->
     case emqx_utils_json:safe_decode(Payload, [return_maps]) of
         {ok, JSON} ->
             {ok, JSON};
@@ -461,7 +461,7 @@ decode(Payload, #{type := json}, Transformation) ->
             },
             {error, TraceFailureContext}
     end;
-decode(Payload, #{type := avro, schema := SerdeName}, Transformation) ->
+decode(Payload, #{type := avro, schema := SerdeName}, Transformation) when is_binary(Payload) ->
     try
         {ok, emqx_schema_registry_serde:decode(SerdeName, Payload)}
     catch
@@ -531,7 +531,26 @@ decode(
                 }
             },
             {error, TraceFailureContext}
-    end.
+    end;
+decode(NotABinary, #{} = Decoder, Transformation) ->
+    DecoderContext0 = maps:with([type, name, message_type], Decoder),
+    DecoderContext1 = emqx_utils_maps:rename(name, schema_name, DecoderContext0),
+    DecoderContext = emqx_utils_maps:rename(type, decoder, DecoderContext1),
+    Context =
+        maps:merge(
+            DecoderContext,
+            #{
+                reason => <<"payload must be a binary">>,
+                hint => <<"check the transformation(s) before this one for inconsistencies">>,
+                bad_payload => NotABinary
+            }
+        ),
+    TraceFailureContext = #trace_failure_context{
+        transformation = Transformation,
+        tag = "payload_decode_failed",
+        context = Context
+    },
+    {error, TraceFailureContext}.
 
 encode(Payload, #{type := none}, _Transformation) ->
     {ok, Payload};

apps/emqx_message_transformation/test/emqx_message_transformation_http_api_SUITE.erl

@@ -2000,3 +2000,44 @@ t_dryrun_transformation(_Config) ->
         []
     ),
     ok.
+
+%% Verifies that if a transformation's decoder is fed a non-binary input (e.g.:
+%% `undefined'), it returns a friendly message.
+t_non_binary_input_for_decoder(_Config) ->
+    ?check_trace(
+        begin
+            %% The transformations set up here lead to an invalid input payload for the
+            %% second transformation: `payload = undefined' (an atom) after the first
+            %% transformation.
+            Name1 = <<"foo">>,
+            Operations1 = [operation(payload, <<"flags.dup">>)],
+            NoSerde = #{<<"type">> => <<"none">>},
+            Transformation1 = transformation(Name1, Operations1, #{
+                <<"payload_decoder">> => NoSerde,
+                <<"payload_encoder">> => NoSerde
+            }),
+            Name2 = <<"bar">>,
+            Operations2 = [],
+            JSONSerde = #{<<"type">> => <<"json">>},
+            Transformation2 = transformation(Name2, Operations2, #{
+                <<"payload_decoder">> => JSONSerde,
+                <<"payload_encoder">> => JSONSerde
+            }),
+            {201, _} = insert(Transformation1),
+            {201, _} = insert(Transformation2),
+
+            C = connect(<<"c1">>),
+            {ok, _, [_]} = emqtt:subscribe(C, <<"t/#">>),
+            ok = publish(C, <<"t/1">>, #{x => 1, y => true}),
+            ?assertNotReceive({publish, _}),
+
+            ok
+        end,
+        fun(Trace) ->
+            SubTrace = ?of_kind(message_transformation_failed, Trace),
+            ?assertMatch([], [E || E = #{reason := function_clause} <- SubTrace]),
+            ?assertMatch([#{reason := <<"payload must be a binary">>} | _], SubTrace),
+            ok
+        end
+    ),
+    ok.

changes/ee/fix-13735.en.md

@@ -0,0 +1 @@
+Improved message transformation error messages when payload to be decoded is invalid.