fix(emqx_bridge): handle errors from pre/post_config_update

apps/emqx_bridge/src/emqx_bridge_v2_api.erl

@@ -747,7 +747,12 @@ create_or_update_bridge(BridgeType, BridgeName, Conf, HttpStatusCode) ->
     case emqx_bridge_v2:create(BridgeType, BridgeName, Conf) of
         {ok, _} ->
             lookup_from_all_nodes(BridgeType, BridgeName, HttpStatusCode);
-        {error, Reason} when is_map(Reason) ->
+        {error, {PreOrPostConfigUpdate, _HandlerMod, Reason}} when
+            PreOrPostConfigUpdate =:= pre_config_update;
+            PreOrPostConfigUpdate =:= post_config_update
+        ->
+            ?BAD_REQUEST(map_to_json(redact(Reason)));
+        {error, Reason} ->
             ?BAD_REQUEST(map_to_json(redact(Reason)))
     end.
 

apps/emqx_bridge/test/emqx_bridge_v2_api_SUITE.erl

@@ -100,6 +100,11 @@
 }).
 -define(KAFKA_BRIDGE(Name), ?KAFKA_BRIDGE(Name, ?CONNECTOR_NAME)).
 
+-define(KAFKA_BRIDGE_UPDATE(Name, Connector),
+    maps:without([<<"name">>, <<"type">>], ?KAFKA_BRIDGE(Name, Connector))
+).
+-define(KAFKA_BRIDGE_UPDATE(Name), ?KAFKA_BRIDGE_UPDATE(Name, ?CONNECTOR_NAME)).
+
 %% -define(BRIDGE_TYPE_MQTT, <<"mqtt">>).
 %% -define(MQTT_BRIDGE(SERVER, NAME), ?BRIDGE(NAME, ?BRIDGE_TYPE_MQTT)#{
 %%     <<"server">> => SERVER,
@@ -395,18 +400,102 @@ t_bridges_lifecycle(Config) ->
         request_json(
             put,
             uri([?ROOT, BridgeID]),
-            maps:without(
-                [<<"type">>, <<"name">>],
-                ?KAFKA_BRIDGE(?BRIDGE_NAME, <<"foobla">>)
-            ),
+            ?KAFKA_BRIDGE_UPDATE(?BRIDGE_NAME, <<"foobla">>),
             Config
         )
     ),
 
+    %% update bridge with unknown connector name
+    {ok, 400, #{
+        <<"code">> := <<"BAD_REQUEST">>,
+        <<"message">> := Message1
+    }} =
+        request_json(
+            put,
+            uri([?ROOT, BridgeID]),
+            ?KAFKA_BRIDGE_UPDATE(?BRIDGE_NAME, <<"does_not_exist">>),
+            Config
+        ),
+    ?assertMatch(
+        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
+        emqx_utils_json:decode(Message1)
+    ),
+
+    %% update bridge with connector of wrong type
+    {ok, 201, _} =
+        request(
+            post,
+            uri(["connectors"]),
+            (?CONNECTOR(<<"foobla2">>))#{
+                <<"type">> => <<"azure_event_hub_producer">>,
+                <<"authentication">> => #{
+                    <<"username">> => <<"emqxuser">>,
+                    <<"password">> => <<"topSecret">>,
+                    <<"mechanism">> => <<"plain">>
+                },
+                <<"ssl">> => #{
+                    <<"enable">> => true,
+                    <<"server_name_indication">> => <<"auto">>,
+                    <<"verify">> => <<"verify_none">>,
+                    <<"versions">> => [<<"tlsv1.3">>, <<"tlsv1.2">>]
+                }
+            },
+            Config
+        ),
+    {ok, 400, #{
+        <<"code">> := <<"BAD_REQUEST">>,
+        <<"message">> := Message2
+    }} =
+        request_json(
+            put,
+            uri([?ROOT, BridgeID]),
+            ?KAFKA_BRIDGE_UPDATE(?BRIDGE_NAME, <<"foobla2">>),
+            Config
+        ),
+    ?assertMatch(
+        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
+        emqx_utils_json:decode(Message2)
+    ),
+
     %% delete the bridge
     {ok, 204, <<>>} = request(delete, uri([?ROOT, BridgeID]), Config),
     {ok, 200, []} = request_json(get, uri([?ROOT]), Config),
 
+    %% try create with unknown connector name
+    {ok, 400, #{
+        <<"code">> := <<"BAD_REQUEST">>,
+        <<"message">> := Message3
+    }} =
+        request_json(
+            post,
+            uri([?ROOT]),
+            ?KAFKA_BRIDGE(?BRIDGE_NAME, <<"does_not_exist">>),
+            Config
+        ),
+    ?assertMatch(
+        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
+        emqx_utils_json:decode(Message3)
+    ),
+
+    %% try create bridge with connector of wrong type
+    {ok, 400, #{
+        <<"code">> := <<"BAD_REQUEST">>,
+        <<"message">> := Message4
+    }} =
+        request_json(
+            post,
+            uri([?ROOT]),
+            ?KAFKA_BRIDGE(?BRIDGE_NAME, <<"foobla2">>),
+            Config
+        ),
+    ?assertMatch(
+        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
+        emqx_utils_json:decode(Message4)
+    ),
+
+    %% make sure nothing has been created above
+    {ok, 200, []} = request_json(get, uri([?ROOT]), Config),
+
     %% update a deleted bridge returns an error
     ?assertMatch(
         {ok, 404, #{
@@ -416,15 +505,12 @@ t_bridges_lifecycle(Config) ->
         request_json(
             put,
             uri([?ROOT, BridgeID]),
-            maps:without(
-                [<<"type">>, <<"name">>],
-                ?KAFKA_BRIDGE(?BRIDGE_NAME)
-            ),
+            ?KAFKA_BRIDGE_UPDATE(?BRIDGE_NAME),
             Config
         )
     ),
 
-    %% Deleting a non-existing bridge should result in an error
+    %% deleting a non-existing bridge should result in an error
     ?assertMatch(
         {ok, 404, #{
             <<"code">> := <<"NOT_FOUND">>,