fix: return friendly message when kafka producer fails to start (rv5.0)

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

The returned information does not allow to diagnose the issue (i.e.: a
connection issue due to the wrong host and port, the wrong password
failing authn).  However, such information is printed to the logs.

This changes the returned error to the API so that the user is hinted
at looking at the logs for further investigation of the error.

apps/emqx_resource/src/emqx_resource.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_resource, [
     {description, "Manager for all external resources"},
-    {vsn, "0.1.10"},
+    {vsn, "0.1.11"},
     {registered, []},
     {mod, {emqx_resource_app, []}},
     {applications, [

apps/emqx_resource/src/emqx_resource.erl

@@ -356,7 +356,14 @@ is_buffer_supported(Module) ->
 -spec call_start(manager_id(), module(), resource_config()) ->
     {ok, resource_state()} | {error, Reason :: term()}.
 call_start(MgrId, Mod, Config) ->
-    ?SAFE_CALL(Mod:on_start(MgrId, Config)).
+    try
+        Mod:on_start(MgrId, Config)
+    catch
+        throw:{error, Error} ->
+            {error, Error};
+        Kind:Error:Stacktrace ->
+            {error, {Kind, Error, Stacktrace}}
+    end.
 
 -spec call_health_check(manager_id(), module(), resource_state()) ->
     resource_status()

lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_consumer.erl

@@ -152,7 +152,11 @@ on_start(InstanceId, Config) ->
                 kafka_hosts => BootstrapHosts,
                 reason => emqx_misc:redact(Reason)
             }),
-            throw(failed_to_start_kafka_client)
+            throw(
+                {error,
+                    "Failed to start Kafka client. Please check the logs for errors and check"
+                    " the connection parameters"}
+            )
     end,
     start_consumer(Config, InstanceId, ClientID).
 

lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl

@@ -114,7 +114,11 @@ on_start(InstId, Config) ->
                     client_id => ClientId
                 }
             ),
-            throw(failed_to_start_kafka_producer)
+            throw(
+                {error,
+                    "Failed to start Kafka client. Please check the logs for errors and check"
+                    " the connection parameters"}
+            )
     end.
 
 on_stop(_InstanceID, #{client_id := ClientID, producers := Producers, resource_id := ResourceID}) ->

lib-ee/emqx_ee_bridge/test/emqx_bridge_impl_kafka_producer_SUITE.erl

@@ -9,6 +9,7 @@
 
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 -include_lib("brod/include/brod.hrl").
 
 -define(PRODUCER, emqx_bridge_impl_kafka_producer).
@@ -415,9 +416,14 @@ t_failed_creation_then_fix(Config) ->
         Type, erlang:list_to_atom(Name), WrongConf
     ),
     WrongConfigAtom = WrongConfigAtom1#{bridge_name => Name},
-    ?assertThrow(failed_to_start_kafka_producer, ?PRODUCER:on_start(ResourceId, WrongConfigAtom)),
-    %% before throwing, it should cleanup the client process.
-    ?assertEqual([], supervisor:which_children(wolff_client_sup)),
+    ?assertThrow(
+        {error, _},
+        ?PRODUCER:on_start(ResourceId, WrongConfigAtom)
+    ),
+    %% before throwing, it should cleanup the client process.  we
+    %% retry because the supervisor might need some time to really
+    %% remove it from its tree.
+    ?retry(50, 10, ?assertEqual([], supervisor:which_children(wolff_client_sup))),
     %% must succeed with correct config
     {ok, #{config := ValidConfigAtom1}} = emqx_bridge:create(
         Type, erlang:list_to_atom(Name), ValidConf