Merge pull request #8697 from rabbitmq/rabbitmq-server-8692

Gracefully close shovel if destination exchange does not exist (#8692 with some cosmetic changes)

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -263,7 +263,7 @@ handle_source({'EXIT', Conn, Reason},
               #{source := #{current := {Conn, _, _}}}) ->
     {stop, {inbound_conn_died, Reason}};
 
-handle_source({'EXIT', _Pid, {shutdown, {server_initiated_close, ?PRECONDITION_FAILED, Reason}}}, _State) ->
+handle_source({'EXIT', _Pid, {shutdown, {server_initiated_close, _, Reason}}}, _State) ->
     {stop, {inbound_link_or_channel_closure, Reason}};
 
 handle_source(_Msg, _State) ->
@@ -288,7 +288,7 @@ handle_dest(#'basic.cancel'{}, #{name := Name}) ->
 handle_dest({'EXIT', Conn, Reason}, #{dest := #{current := {Conn, _, _}}}) ->
     {stop, {outbound_conn_died, Reason}};
 
-handle_dest({'EXIT', _Pid, {shutdown, {server_initiated_close, ?PRECONDITION_FAILED, Reason}}}, _State) ->
+handle_dest({'EXIT', _Pid, {shutdown, {server_initiated_close, _, Reason}}}, _State) ->
     {stop, {outbound_link_or_channel_closure, Reason}};
 
 handle_dest(#'connection.blocked'{}, State) ->

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -202,7 +202,7 @@ handle_source({'EXIT', Conn, Reason},
               #{source := #{current := #{conn := Conn}}}) ->
     {stop, {outbound_conn_died, Reason}};
 
-handle_source({'EXIT', _Pid, {shutdown, {server_initiated_close, ?PRECONDITION_FAILED, Reason}}}, _State) ->
+handle_source({'EXIT', _Pid, {shutdown, {server_initiated_close, _, Reason}}}, _State) ->
     {stop, {inbound_link_or_channel_closure, Reason}};
 
 handle_source(_Msg, _State) ->
@@ -260,7 +260,7 @@ handle_dest({'EXIT', Conn, Reason},
             #{dest := #{current := #{conn := Conn}}}) ->
     {stop, {outbound_conn_died, Reason}};
 
-handle_dest({'EXIT', _Pid, {shutdown, {server_initiated_close, ?PRECONDITION_FAILED, Reason}}}, _State) ->
+handle_dest({'EXIT', _Pid, {shutdown, {server_initiated_close, _, Reason}}}, _State) ->
     {stop, {outbound_link_or_channel_closure, Reason}};
 
 handle_dest(_Msg, _State) ->

deps/rabbitmq_shovel/test/dynamic_SUITE.erl

@@ -31,6 +31,7 @@ groups() ->
           set_empty_properties_using_map,
           headers,
           exchange,
+          missing_dest_exchange,
           restart,
           change_definition,
           autodelete,
@@ -262,6 +263,36 @@ exchange(Config) ->
                              <<"queue">>, <<"hello">>)
       end).
 
+
+missing_dest_exchange(Config) ->
+    with_ch(Config,
+        fun (Ch) ->
+            amqp_channel:call(
+            Ch, #'queue.declare'{queue = <<"src">>,
+                                 durable = true}),
+            amqp_channel:call(
+              Ch, #'queue.declare'{queue = <<"dest">>,
+                                   durable = true}),
+            amqp_channel:call(
+              Ch, #'queue.bind'{queue = <<"src">>,
+                                exchange = <<"amq.direct">>,
+                                routing_key = <<"src-key">>}),
+            shovel_test_utils:set_param(Config,
+                        <<"test">>, [{<<"src-queue">>, <<"src">>},
+                                        {<<"dest-exchange">>, <<"dest-ex">>},
+                                        {<<"dest-exchange-key">>, <<"dest-key">>},
+                                        {<<"src-prefetch-count">>, 1}]),
+            publish(Ch, <<"amq.direct">>, <<"src-key">>, <<"hello">>),
+            expect_empty(Ch, <<"src">>),
+            amqp_channel:call(
+              Ch, #'exchange.declare'{exchange = <<"dest-ex">>}),
+            amqp_channel:call(
+              Ch, #'queue.bind'{queue = <<"dest">>,
+                                exchange = <<"dest-ex">>,
+                                routing_key = <<"dest-key">>}),
+            publish_expect(Ch, <<"amq.direct">>, <<"src-key">>, <<"dest">>, <<"hello!">>)
+end).
+
 restart(Config) ->
     with_ch(Config,
       fun (Ch) ->