Merge pull request #14757 from rabbitmq/shovel-revert-fix

Shovels bugfixes

Author: michaelklishin

deps/amqp10_client/src/amqp10_client_connection.erl

@@ -352,6 +352,9 @@ close_sent(_EvtType, heartbeat, _Data) ->
 close_sent(_EvtType, {'EXIT', _Pid, shutdown}, _Data) ->
     %% monitored processes may exit during closure
     keep_state_and_data;
+close_sent(_EvtType, {'EXIT', _Pid, {shutdown, _}}, _Data) ->
+    %% monitored processes may exit during closure
+    keep_state_and_data;
 close_sent(_EvtType, {'DOWN', _Ref, process, ReaderPid, _Reason},
            #state{reader = ReaderPid}) ->
     %% if the reader exits we probably won't receive a close frame

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -306,13 +306,13 @@ ack(Tag, false, State = #{source := #{current := #{link := LinkRef}} = Src}) ->
 -spec nack(Tag :: tag(), Multi :: boolean(), state()) -> state().
 nack(Tag, false, State = #{source := #{current := #{link := LinkRef}} = Src}) ->
     % the tag is the same as the deliveryid
-    ok = amqp10_client_session:disposition(LinkRef, Tag, Tag, true, released),
-    State#{source => Src#{last_nacked_tag => Tag}};
+    ok = amqp10_client_session:disposition(LinkRef, Tag, Tag, true, rejected),
+    State#{source => Src#{last_acked_tag => Tag}};
 nack(Tag, true, State = #{source := #{current := #{link := LinkRef},
-                                      last_nacked_tag := LastTag} = Src}) ->
+                                      last_acked_tag := LastTag} = Src}) ->
     First = LastTag + 1,
-    ok = amqp10_client_session:disposition(LinkRef, First, Tag, true, released),
-    State#{source => Src#{last_nacked_tag => Tag}}.
+    ok = amqp10_client_session:disposition(LinkRef, First, Tag, true, rejected),
+    State#{source => Src#{last_acked_tag => Tag}}.
 
 status(#{dest := #{current := #{link_state := attached}}}) ->
     flow;

deps/rabbitmq_shovel/test/shovel_dynamic_SUITE.erl

@@ -71,8 +71,12 @@ tests() ->
      autodelete_quorum_on_confirm_no_transfer,
      autodelete_classic_on_publish_no_transfer,
      autodelete_quorum_on_publish_no_transfer,
-     autodelete_classic_on_confirm_with_rejections,
-     autodelete_quorum_on_confirm_with_rejections,
+     %% AMQP091 and local shovels requeue messages on reject
+     %% AMQP10 discards messages on reject
+     %% These two tests will remain commented out until the
+     %% behaviour is unified.
+     %% autodelete_classic_on_confirm_with_rejections,
+     %% autodelete_quorum_on_confirm_with_rejections,
      autodelete_classic_on_publish_with_rejections,
      autodelete_quorum_on_publish_with_rejections
     ].