Change modified outcome behaviour

With the new quorum queue v4 improvements where a requeue counter was
added in addition to the quorum queue delivery counter, the following
sentence from #6292 (comment)
doesn't apply anymore:

> Also the case where delivery_failed=false|undefined requires the release of the
> message without incrementing the delivery_count. Again this is not something
> that our queues are able to do so again we have to reject without requeue.

Therefore, we simplify the modified outcome behaviour:
RabbitMQ will from now on only discard the message if the modified's
undeliverable-here field is true.

Author: ansd

deps/amqp10_client/src/amqp10_msg.erl

@@ -193,7 +193,8 @@ header(first_acquirer = K,
 header(delivery_count = K,
        #amqp10_msg{header = #'v1_0.header'{delivery_count = D}}) ->
     header_value(K, D);
-header(K, #amqp10_msg{header = undefined}) -> header_value(K, undefined).
+header(K, #amqp10_msg{header = undefined}) ->
+    header_value(K, undefined).
 
 -spec delivery_annotations(amqp10_msg()) -> #{annotations_key() => any()}.
 delivery_annotations(#amqp10_msg{delivery_annotations = undefined}) ->

deps/rabbit/src/rabbit_amqp_session.erl

@@ -1861,20 +1861,21 @@ settle_op_from_outcome(#'v1_0.rejected'{}) ->
     discard;
 settle_op_from_outcome(#'v1_0.released'{}) ->
     requeue;
-%% Keep the same Modified behaviour as in RabbitMQ 3.x
-settle_op_from_outcome(#'v1_0.modified'{delivery_failed = true,
-                                        undeliverable_here = UndelHere})
-  when UndelHere =/= true ->
-    requeue;
-settle_op_from_outcome(#'v1_0.modified'{}) ->
-    %% If delivery_failed is not true, we can't increment its delivery_count.
-    %% So, we will have to reject without requeue.
-    %%
-    %% If undeliverable_here is true, this is not quite correct because
-    %% undeliverable_here refers to the link, and not the message in general.
-    %% However, we cannot filter messages from being assigned to individual consumers.
-    %% That's why we will have to reject it without requeue.
+
+%% RabbitMQ does not support any of the modified outcome fields correctly.
+%% However, we still allow the client to settle with the modified outcome
+%% because some client libraries such as Apache QPid make use of it:
+%% https://github.com/apache/qpid-jms/blob/90eb60f59cb59b7b9ad8363ee8a843d6903b8e77/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java#L464
+%% In such cases, it's better when RabbitMQ does not end the session.
+%% See https://github.com/rabbitmq/rabbitmq-server/issues/6121
+settle_op_from_outcome(#'v1_0.modified'{undeliverable_here = true}) ->
+    %% This is not quite correct because undeliverable_here refers to the link,
+    %% and not the message in general. However, RabbitMQ cannot filter messages from
+    %% being assigned to individual consumers. That's why we discard.
     discard;
+settle_op_from_outcome(#'v1_0.modified'{}) ->
+    requeue;
+
 settle_op_from_outcome(Outcome) ->
     protocol_error(
       ?V_1_0_AMQP_ERROR_INVALID_FIELD,

deps/rabbit/test/amqp_client_SUITE.erl

@@ -41,6 +41,7 @@ groups() ->
       [
        reliable_send_receive_with_outcomes_classic_queue,
        reliable_send_receive_with_outcomes_quorum_queue,
+       modified,
        sender_settle_mode_unsettled,
        sender_settle_mode_unsettled_fanout,
        sender_settle_mode_mixed,
@@ -402,6 +403,56 @@ reliable_send_receive(QType, Outcome, Config) ->
     ok = end_session_sync(Session2),
     ok = amqp10_client:close_connection(Connection2).
 
+%% This test case doesn't expect the correct AMQP spec behavivour.
+%% We know that RabbitMQ doesn't implement the modified outcome correctly.
+%% Here, we test RabbitMQ's workaround behaviour:
+%% RabbitMQ discards if undeliverable-here is true. Otherwise, RabbitMQ requeues.
+modified(Config) ->
+    QName = atom_to_binary(?FUNCTION_NAME),
+    {Connection, Session, LinkPair} = init(Config),
+    {ok, #{type := <<"quorum">>}} = rabbitmq_amqp_client:declare_queue(
+                                      LinkPair, QName,
+                                      #{arguments => #{<<"x-queue-type">> => {utf8, <<"quorum">>}}}),
+    Address = rabbitmq_amqp_address:queue(QName),
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address),
+    ok = wait_for_credit(Sender),
+
+    Msg1 = amqp10_msg:new(<<"tag1">>, <<"m1">>, true),
+    Msg2 = amqp10_msg:new(<<"tag2">>, <<"m2">>, true),
+    ok = amqp10_client:send_msg(Sender, Msg1),
+    ok = amqp10_client:send_msg(Sender, Msg2),
+    ok = amqp10_client:detach_link(Sender),
+
+    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, Address, unsettled),
+
+    {ok, M1} = amqp10_client:get_msg(Receiver),
+    ?assertEqual([<<"m1">>], amqp10_msg:body(M1)),
+    ok = amqp10_client:settle_msg(Receiver, M1, {modified, false, _UndeliverableHere = true, #{}}),
+
+    {ok, M2a} = amqp10_client:get_msg(Receiver),
+    ?assertEqual([<<"m2">>], amqp10_msg:body(M2a)),
+    ok = amqp10_client:settle_msg(Receiver, M2a, {modified, false, false, #{}}),
+
+    {ok, M2b} = amqp10_client:get_msg(Receiver),
+    ?assertEqual([<<"m2">>], amqp10_msg:body(M2b)),
+    ok = amqp10_client:settle_msg(Receiver, M2b, {modified, true, false, #{}}),
+
+    {ok, M2c} = amqp10_client:get_msg(Receiver),
+    ?assertEqual([<<"m2">>], amqp10_msg:body(M2c)),
+    ok = amqp10_client:settle_msg(Receiver, M2c, {modified, true, false, #{<<"key">> => <<"val">>}}),
+
+    {ok, M2d} = amqp10_client:get_msg(Receiver),
+    ?assertEqual([<<"m2">>], amqp10_msg:body(M2d)),
+    ?assertEqual(0, amqp10_msg:header(delivery_count, M2d)),
+    ok = amqp10_client:settle_msg(Receiver, M2d, modified),
+
+    ok = amqp10_client:detach_link(Receiver),
+    ?assertMatch({ok, #{message_count := 1}},
+                 rabbitmq_amqp_client:delete_queue(LinkPair, QName)),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
+    ok = end_session_sync(Session),
+    ok = amqp10_client:close_connection(Connection).
+
 %% Tests that confirmations are returned correctly
 %% when sending many messages async to a quorum queue.
 sender_settle_mode_unsettled(Config) ->