fix shovel error according to consumer_timeout

3 files changed, 20 insertions, 0 deletions

diff --git a/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl b/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl
index 9b15d50691..217fad7a3c 100644
--- a/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl
+++ b/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl
@@ -230,6 +230,9 @@ 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) ->
+    {stop, {inbound_consumer_no_ack, Reason}};
+
 handle_source(_Msg, _State) ->
     not_handled.
 
@@ -251,6 +254,9 @@ 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) ->
+    {stop, {outbound_consumer_no_ack, Reason}};
 
 handle_dest(_Msg, _State) ->
     not_handled.
diff --git a/deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl b/deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl
index b4414d18dd..8568a2000e 100644
--- a/deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl
+++ b/deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl
@@ -200,6 +200,10 @@ handle_source({amqp10_event, {link, Link, _Evt}},
 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) ->
+    {stop, {inbound_consumer_no_ack, Reason}};
+    
 handle_source(_Msg, _State) ->
     not_handled.
 
@@ -254,6 +258,10 @@ handle_dest({amqp10_event, {link, Link, _Evt}},
 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) ->
+    {stop, {outbound_consumer_no_ack, Reason}};
+    
 handle_dest(_Msg, _State) ->
     not_handled.
 
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl
index 79c348ec81..e8945cc0bc 100644
--- a/deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl
+++ b/deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl
@@ -113,6 +113,9 @@ handle_info(Msg, State = #state{config = Config, name = Name}) ->
                 {stop, {outbound_conn_died, Reason}} ->
                     rabbit_log_shovel:error("Shovel ~s detected destination connection failure: ~p", [human_readable_name(Name), Reason]),
                     {stop, Reason, State};
+	            {stop, {outbound_consumer_no_ack, Reason}} ->
+    		        rabbit_log_shovel:error("Shovel ~s detected destination shovel failure: ~p", [human_readable_name(Name), Reason]),
+    		        {stop, Reason, State};
                 {stop, Reason} ->
                     rabbit_log_shovel:debug("Shovel ~s decided to stop due a message from destination: ~p", [human_readable_name(Name), Reason]),
                     {stop, Reason, State};
@@ -125,6 +128,9 @@ handle_info(Msg, State = #state{config = Config, name = Name}) ->
         {stop, {inbound_conn_died, Reason}} ->
             rabbit_log_shovel:error("Shovel ~s detected source connection failure: ~p", [human_readable_name(Name), Reason]),
             {stop, Reason, State};
+        {stop, {inbound_consumer_no_ack, Reason}} ->
+	        rabbit_log_shovel:error("Shovel ~s detected source shovel failure: ~p", [human_readable_name(Name), Reason]),
+	        {stop, Reason, State};
         {stop, Reason} ->
             rabbit_log_shovel:error("Shovel ~s decided to stop due a message from source: ~p", [human_readable_name(Name), Reason]),
             {stop, Reason, State};