fix(mqtt_source): don't attempt local publish when legacy config is absent

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

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_ingress.erl

@@ -18,6 +18,7 @@
 
 -include_lib("emqx/include/logger.hrl").
 -include_lib("emqx/include/emqx_mqtt.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 %% management APIs
 -export([
@@ -284,9 +285,10 @@ maybe_on_message_received(Msg, {Mod, Func, Args}) ->
 maybe_on_message_received(_Msg, undefined) ->
     ok.
 
-maybe_publish_local(Msg, Local = #{}, Props) ->
+maybe_publish_local(Msg, Local = #{topic := Topic}, Props) when Topic =/= undefined ->
+    ?tp(mqtt_ingress_publish_local, #{msg => Msg, local => Local}),
     emqx_broker:publish(to_broker_msg(Msg, Local, Props));
-maybe_publish_local(_Msg, undefined, _Props) ->
+maybe_publish_local(_Msg, _Local, _Props) ->
     ok.
 
 %%

apps/emqx_bridge_mqtt/test/emqx_bridge_mqtt_v2_subscriber_SUITE.erl

@@ -238,6 +238,9 @@ t_receive_via_rule(Config) ->
         end,
         fun(Trace) ->
             ?assertEqual([], ?of_kind("action_references_nonexistent_bridges", Trace)),
+            %% We don't have the hidden, legacy `local' config set, so we shouldn't
+            %% attempt to publish directly.
+            ?assertEqual([], ?of_kind(mqtt_ingress_publish_local, Trace)),
             ok
         end
     ),