Log errors when policy change notifications fail (backport #15525)

deps/rabbit/src/rabbit_policy.erl

@@ -413,9 +413,27 @@ update_matched_objects(VHost, PolicyDef, ActingUser) ->
                                         get_updated_exchange(Policies, OpPolicies),
                                         get_updated_queue(Policies, OpPolicies))
         end,
-    _ = [catch maybe_notify_of_policy_change(XRes, PolicyDef, ActingUser) || XRes <- XUpdateResults],
-    _ = [catch maybe_notify_of_policy_change(QRes, PolicyDef, ActingUser) || QRes <- QUpdateResults],
-    ok.
+    lists:foreach(
+      fun(XRes) ->
+              try maybe_notify_of_policy_change(XRes, PolicyDef, ActingUser)
+              catch Class:Reason ->
+                        {#exchange{name = XName}, _} = XRes,
+                        ?LOG_WARNING("Failed to apply policy change "
+                                     "notification for ~ts: ~ts ~tp",
+                                     [rabbit_misc:rs(XName), Class, Reason])
+              end
+      end, XUpdateResults),
+    lists:foreach(
+      fun(QRes) ->
+              try maybe_notify_of_policy_change(QRes, PolicyDef, ActingUser)
+              catch Class:Reason ->
+                        {Q, _} = QRes,
+                        QName = amqqueue:get_name(Q),
+                        ?LOG_WARNING("Failed to apply policy change "
+                                     "notification for ~ts: ~ts ~tp",
+                                     [rabbit_misc:rs(QName), Class, Reason])
+              end
+      end, QUpdateResults).
 
 get_updated_exchange(Policies, OpPolicies) ->
     fun(X = #exchange{name = XName,

deps/rabbitmq_queue_federation/src/rabbit_federation_queue.erl

@@ -17,6 +17,7 @@
                     {enables, recovery}]}).
 
 -include_lib("rabbit/include/amqqueue.hrl").
+-include_lib("kernel/include/logger.hrl").
 -include("rabbit_queue_federation.hrl").
 
 -behaviour(rabbit_queue_decorator).
@@ -47,8 +48,15 @@ policy_changed(Q1, Q2) when ?is_amqqueue(Q1) ->
     QName = amqqueue:get_name(Q1),
     case rabbit_amqqueue:lookup(QName) of
         {ok, Q0} when ?is_amqqueue(Q0) ->
-            rpc:call(amqqueue:qnode(Q0), rabbit_federation_queue,
-                     policy_changed_local, [Q1, Q2]);
+            Node = amqqueue:qnode(Q0),
+            try erpc:call(Node, rabbit_federation_queue,
+                          policy_changed_local, [Q1, Q2])
+            catch Class:Reason ->
+                      ?LOG_WARNING("Queue federation failed to apply policy "
+                                   "change for ~ts on node ~tp: ~s ~tp",
+                                   [rabbit_misc:rs(QName), Node, Class, Reason]),
+                      ok
+            end;
         {error, not_found} ->
             ok
     end.