Revert "Settle unroutable message with released state (backport #8015) (backport #8736)"

Author: michaelklishin

deps/amqp_client/src/amqp_channel.erl

@@ -886,9 +886,6 @@ flush_writer(#state{driver = direct}) ->
     ok.
 amqp_msg(none) ->
     none;
-amqp_msg({DTag, Content}) ->
-    {Props, Payload} = rabbit_basic_common:from_content(Content),
-    {DTag, #amqp_msg{props = Props, payload = Payload}};
 amqp_msg(Content) ->
     {Props, Payload} = rabbit_basic_common:from_content(Content),
     #amqp_msg{props = Props, payload = Payload}.

deps/rabbit/src/rabbit_channel.erl

@@ -113,10 +113,7 @@
           consumer_timeout,
           authz_context,
           %% defines how ofter gc will be executed
-          writer_gc_threshold,
-          %% true with AMQP 1.0 to include the publishing sequence
-          %% in the return callback, false otherwise
-          extended_return_callback
+          writer_gc_threshold
          }).
 
 -record(pending_ack, {
@@ -523,7 +520,6 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
     MaxMessageSize = get_max_message_size(),
     ConsumerTimeout = get_consumer_timeout(),
     OptionalVariables = extract_variable_map_from_amqp_params(AmqpParams),
-    UseExtendedReturnCallback = use_extended_return_callback(AmqpParams),
     {ok, GCThreshold} = application:get_env(rabbit, writer_gc_threshold),
     State = #ch{cfg = #conf{state = starting,
                             protocol = Protocol,
@@ -542,8 +538,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                             max_message_size = MaxMessageSize,
                             consumer_timeout = ConsumerTimeout,
                             authz_context = OptionalVariables,
-                            writer_gc_threshold = GCThreshold,
-                            extended_return_callback = UseExtendedReturnCallback
+                            writer_gc_threshold = GCThreshold
                            },
                 limiter = Limiter,
                 tx                      = none,
@@ -1106,15 +1101,6 @@ extract_variable_map_from_amqp_params([Value]) ->
 extract_variable_map_from_amqp_params(_) ->
     #{}.
 
-%% Use tuple representation of amqp_params to avoid a dependency on amqp_client.
-%% Used for AMQP 1.0
-use_extended_return_callback({amqp_params_direct,_,_,_,_,
-                              {amqp_adapter_info,_,_,_,_,_,{'AMQP',"1.0"},_},
-                              _}) ->
-    true;
-use_extended_return_callback(_) ->
-    false.
-
 check_msg_size(Content, MaxMessageSize, GCThreshold) ->
     Size = rabbit_basic:maybe_gc_large_msg(Content, GCThreshold),
     case Size of
@@ -1956,8 +1942,9 @@ binding_action(Fun, SourceNameBin0, DestinationType, DestinationNameBin0,
             ok
     end.
 
-basic_return(Content, #basic_message{exchange_name = ExchangeName,
-                                     routing_keys  = [RoutingKey | _CcRoutes]},
+basic_return(#basic_message{exchange_name = ExchangeName,
+                            routing_keys  = [RoutingKey | _CcRoutes],
+                            content       = Content},
              State = #ch{cfg = #conf{protocol = Protocol,
                                      writer_pid = WriterPid}},
              Reason) ->
@@ -2193,9 +2180,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
                                         mandatory  = Mandatory,
                                         confirm    = Confirm,
                                         msg_seq_no = MsgSeqNo},
-                                        RoutedToQueueNames = [QName]},
-                                        State0 = #ch{cfg = #conf{extended_return_callback = ExtendedReturnCallback},
-                                                     queue_states = QueueStates0}) -> %% optimisation when there is one queue
+                   RoutedToQueueNames = [QName]}, State0 = #ch{queue_states = QueueStates0}) -> %% optimisation when there is one queue
     Qs0 = rabbit_amqqueue:lookup(RoutedToQueueNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     QueueNames = lists:map(fun amqqueue:get_name/1, Qs),
@@ -2204,7 +2189,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
             rabbit_global_counters:messages_routed(amqp091, erlang:min(1, length(Qs))),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
-            ok = process_routing_mandatory(ExtendedReturnCallback, Mandatory, Qs, MsgSeqNo, Message, State0),
+            ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
             State1 = process_routing_confirm(Confirm, QueueNames, MsgSeqNo, XName, State0),
             %% Actions must be processed after registering confirms as actions may
             %% contain rejections of publishes
@@ -2232,9 +2217,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
                                         mandatory  = Mandatory,
                                         confirm    = Confirm,
                                         msg_seq_no = MsgSeqNo},
-                   RoutedToQueueNames},
-                   State0 = #ch{cfg = #conf{extended_return_callback = ExtendedReturnCallback},
-                                queue_states = QueueStates0}) ->
+                   RoutedToQueueNames}, State0 = #ch{queue_states = QueueStates0}) ->
     Qs0 = rabbit_amqqueue:lookup(RoutedToQueueNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     QueueNames = lists:map(fun amqqueue:get_name/1, Qs),
@@ -2243,7 +2226,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
             rabbit_global_counters:messages_routed(amqp091, length(Qs)),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
-            ok = process_routing_mandatory(ExtendedReturnCallback, Mandatory, Qs, MsgSeqNo, Message, State0),
+            ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
             State1 = process_routing_confirm(Confirm, QueueNames,
                                              MsgSeqNo, XName, State0),
             %% Actions must be processed after registering confirms as actions may
@@ -2265,32 +2248,19 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
               [rabbit_misc:rs(Resource)])
     end.
 
-process_routing_mandatory(_ExtendedReturnCallback = false,
-                          _Mandatory = true,
-                          _RoutedToQs = [],
-                          _MsgSeqNo,
-                          #basic_message{content = Content} = Msg, State) ->
-    rabbit_global_counters:messages_unroutable_returned(amqp091, 1),
-    ok = basic_return(Content, Msg, State, no_route),
-    ok;
-process_routing_mandatory(_ExtendedReturnCallback = true,
-                          _Mandatory = true,
+process_routing_mandatory(_Mandatory = true,
                           _RoutedToQs = [],
-                          MsgSeqNo,
-                          #basic_message{content = Content} = Msg, State) ->
+                          Msg, State) ->
     rabbit_global_counters:messages_unroutable_returned(amqp091, 1),
-    %% providing the publishing sequence for AMQP 1.0
-    ok = basic_return({MsgSeqNo, Content}, Msg, State, no_route),
+    ok = basic_return(Msg, State, no_route),
     ok;
-process_routing_mandatory(_ExtendedReturnCallback,
-                          _Mandatory = false,
+process_routing_mandatory(_Mandatory = false,
                           _RoutedToQs = [],
-                          _MsgSeqNo,
                           #basic_message{exchange_name = ExchangeName}, State) ->
     rabbit_global_counters:messages_unroutable_dropped(amqp091, 1),
     ?INCR_STATS(exchange_stats, ExchangeName, 1, drop_unroutable, State),
     ok;
-process_routing_mandatory(_, _, _, _, _, _) ->
+process_routing_mandatory(_, _, _, _) ->
     ok.
 
 process_routing_confirm(false, _, _, _, State) ->

deps/rabbit_common/src/rabbit_writer.erl

@@ -105,10 +105,7 @@
 
 -spec send_command(pid(), rabbit_framing:amqp_method_record()) -> 'ok'.
 -spec send_command
-        (pid(), rabbit_framing:amqp_method_record(),
-         rabbit_types:content() |
-         {integer(), rabbit_types:content()} %% publishing sequence for AMQP 1.0 return callback
-        ) ->
+        (pid(), rabbit_framing:amqp_method_record(), rabbit_types:content()) ->
             'ok'.
 -spec send_command_sync(pid(), rabbit_framing:amqp_method_record()) -> 'ok'.
 -spec send_command_sync

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_incoming_link.erl

@@ -17,7 +17,7 @@
 %% Just make these constant for the time being.
 -define(INCOMING_CREDIT, 65536).
 
--record(incoming_link, {name, exchange, routing_key, mandatory,
+-record(incoming_link, {name, exchange, routing_key,
                         delivery_id = undefined,
                         delivery_count = 0,
                         send_settle_mode = undefined,
@@ -53,7 +53,6 @@ attach(#'v1_0.attach'{name = Name,
                            SndSettleMode == ?V_1_0_SENDER_SETTLE_MODE_MIXED ->
                         amqp_channel:register_confirm_handler(BCh, self()),
                         rabbit_amqp1_0_channel:call(BCh, #'confirm.select'{}),
-                        amqp_channel:register_return_handler(BCh, self()),
                         true
                 end,
             Flow = #'v1_0.flow'{ handle = Handle,
@@ -70,8 +69,7 @@ attach(#'v1_0.attach'{name = Name,
               initial_delivery_count = undefined, % must be, I am the receiver
               role = ?RECV_ROLE}, %% server is receiver
             IncomingLink1 =
-                IncomingLink#incoming_link{recv_settle_mode = RcvSettleMode,
-                                           mandatory = Confirm},
+                IncomingLink#incoming_link{recv_settle_mode = RcvSettleMode},
             {ok, [Attach, Flow], IncomingLink1, Confirm};
         {error, Reason} ->
             %% TODO proper link establishment protocol here?
@@ -144,8 +142,7 @@ transfer(#'v1_0.transfer'{delivery_id     = DeliveryId0,
            end,
     rabbit_amqp1_0_channel:cast_flow(
       BCh, #'basic.publish'{exchange    = X,
-                            routing_key = RKey,
-                            mandatory = true}, Msg),
+                            routing_key = RKey}, Msg),
     {SendFlow, CreditUsed1} = case CreditUsed - 1 of
                                   C when C =< 0 ->
                                       {true,  ?INCOMING_CREDIT div 2};
@@ -209,7 +206,6 @@ ensure_target(Target = #'v1_0.target'{address       = Address,
                                     dest, DCh, Dest, DeclareParams,
                                     RouteState)
                           end),
-                    maybe_ensure_queue(Dest, DCh),
                     {XName, RK} = rabbit_routing_util:parse_routing(Dest),
                     {ok, Target, Link#incoming_link{
                                    route_state = RouteState1,
@@ -226,20 +222,6 @@ ensure_target(Target = #'v1_0.target'{address       = Address,
             {error, {address_not_utf8_string, Address}}
     end.
 
-maybe_ensure_queue({amqqueue, Q}, Ch) ->
-    try
-        rabbit_amqp1_0_channel:convert_error(
-          fun () ->
-                  Method = #'queue.declare'{queue = list_to_binary(Q),
-                                            passive = true},
-                  amqp_channel:call(Ch, Method)
-          end)
-    catch exit:#'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED} ->
-              ok
-    end;
-maybe_ensure_queue(_, _) ->
-    ok.
-
 incoming_flow(#incoming_link{ delivery_count = Count }, Handle) ->
     #'v1_0.flow'{handle         = Handle,
                  delivery_count = {uint, Count},

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -14,7 +14,7 @@
          incr_incoming_id/1, next_delivery_id/1, transfers_left/1,
          record_transfers/2, bump_outgoing_window/1,
          record_outgoing/4, settle/3, flow_fields/2, channel/1,
-         flow/2, ack/2, return/2, validate_attach/1]).
+         flow/2, ack/2, validate_attach/1]).
 
 -import(rabbit_amqp1_0_util, [protocol_error/3,
                               serial_add/2, serial_diff/2, serial_compare/2]).
@@ -396,25 +396,3 @@ acknowledgement(DeliveryIds, Disposition) ->
                                     last = {uint, lists:last(DeliveryIds)},
                                     settled = true,
                                     state = #'v1_0.accepted'{} }.
-
-return(DTag, Session = #session{incoming_unsettled_map = Unsettled}) ->
-    {DeliveryId,
-     Unsettled1} = case gb_trees:lookup(DTag, Unsettled) of
-                       {value, #incoming_delivery{ delivery_id = Id }} ->
-                           {Id, gb_trees:delete(DTag, Unsettled)};
-                       none ->
-                           {undefined, Unsettled}
-                   end,
-    Disposition = case DeliveryId of
-                      undefined -> undefined;
-                      _  -> release(DeliveryId,
-                                    #'v1_0.disposition'{role = ?RECV_ROLE})
-    end,
-    {Disposition,
-     Session#session{incoming_unsettled_map = Unsettled1}}.
-
-release(DeliveryId, Disposition) ->
-    Disposition#'v1_0.disposition'{ first = {uint, DeliveryId},
-                                    last = {uint, DeliveryId},
-                                    settled = true,
-                                    state = #'v1_0.released'{} }.

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_process.erl

@@ -130,24 +130,6 @@ handle_info(#'basic.ack'{} = Ack, State = #state{writer_pid = WriterPid,
         F <- rabbit_amqp1_0_session:flow_fields(Reply, Session)],
     {noreply, state(Session1, State)};
 
-handle_info({#'basic.return'{}, {DTag, _Msg}}, State = #state{writer_pid = WriterPid,
-                                                              session    = Session}) ->
-    {Reply, Session1} = rabbit_amqp1_0_session:return(DTag, Session),
-    case Reply of
-        undefined ->
-            ok;
-        _ ->
-            rabbit_amqp1_0_writer:send_command(
-              WriterPid,
-              rabbit_amqp1_0_session:flow_fields(Reply, Session)
-             )
-    end,
-    {noreply, state(Session1, State)};
-
-handle_info({#'basic.return'{}, _Msg}, State = #state{session = Session}) ->
-    rabbit_log:warning("AMQP 1.0 message return without publishing sequence"),
-    {noreply, state(Session, State)};
-
 handle_info({bump_credit, Msg}, State) ->
     credit_flow:handle_bump_msg(Msg),
     {noreply, State};

deps/rabbitmq_amqp1_0/test/amqp10_client_SUITE.erl

@@ -24,8 +24,6 @@ groups() ->
     [
      {tests, [], [
                   reliable_send_receive_with_outcomes,
-                  publishing_to_non_existing_queue_should_settle_with_released,
-                  open_link_to_non_existing_destination_should_end_session,
                   roundtrip_classic_queue_with_drain,
                   roundtrip_quorum_queue_with_drain,
                   roundtrip_stream_queue_with_drain,
@@ -165,68 +163,6 @@ reliable_send_receive(Config, Outcome) ->
 
     ok.
 
-publishing_to_non_existing_queue_should_settle_with_released(Config) ->
-    Container = atom_to_binary(?FUNCTION_NAME, utf8),
-    Suffix = <<"foo">>,
-    %% does not exist
-    QName = <<Container/binary, Suffix/binary>>,
-    Host = ?config(rmq_hostname, Config),
-    Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
-    Address = <<"/exchange/amq.direct/", QName/binary>>,
-
-    OpnConf = #{address => Host,
-                port => Port,
-                container_id => Container,
-                sasl => {plain, <<"guest">>, <<"guest">>}},
-    {ok, Connection} = amqp10_client:open_connection(OpnConf),
-    {ok, Session} = amqp10_client:begin_session(Connection),
-    SenderLinkName = <<"test-sender">>,
-    {ok, Sender} = amqp10_client:attach_sender_link(Session,
-                                                    SenderLinkName,
-                                                    Address),
-    ok = wait_for_credit(Sender),
-    DTag1 = <<"dtag-1">>,
-    %% create an unsettled message,
-    %% link will be in "mixed" mode by default
-    Msg1 = amqp10_msg:new(DTag1, <<"body-1">>, false),
-    ok = amqp10_client:send_msg(Sender, Msg1),
-    ok = wait_for_settlement(DTag1, released),
-
-    ok = amqp10_client:detach_link(Sender),
-    ok = amqp10_client:close_connection(Connection),
-    flush("post sender close"),
-    ok.
-
-open_link_to_non_existing_destination_should_end_session(Config) ->
-    Container = atom_to_list(?FUNCTION_NAME),
-    Name = Container ++ "foo",
-    Addresses = [
-                 "/exchange/" ++ Name ++ "/bar",
-                 "/amq/queue/" ++ Name
-                ],
-    Host = ?config(rmq_hostname, Config),
-    Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
-    OpnConf = #{address => Host,
-                port => Port,
-                container_id => list_to_binary(Container),
-                sasl => {plain, <<"guest">>, <<"guest">>}},
-
-    [begin
-         {ok, Connection} = amqp10_client:open_connection(OpnConf),
-         {ok, Session} = amqp10_client:begin_session(Connection),
-         SenderLinkName = <<"test-sender">>,
-         ct:pal("Address ~p", [Address]),
-         {ok, _} = amqp10_client:attach_sender_link(Session,
-                                                    SenderLinkName,
-                                                    list_to_binary(Address)),
-
-         wait_for_session_end(Session),
-         ok = amqp10_client:close_connection(Connection),
-         flush("post sender close")
-
-     end || Address <- Addresses],
-    ok.
-
 roundtrip_classic_queue_with_drain(Config) ->
     QName  = atom_to_binary(?FUNCTION_NAME, utf8),
     roundtrip_queue_with_drain(Config, <<"classic">>, QName).
@@ -446,27 +382,14 @@ wait_for_credit(Sender) ->
               ct:fail(credited_timeout)
     end.
 
-wait_for_session_end(Session) ->
-    receive
-        {amqp10_event, {session, Session, {ended, _}}} ->
-            flush(?FUNCTION_NAME),
-            ok
-    after 5000 ->
-              flush("wait_for_session_end timed out"),
-              ct:fail(settled_timeout)
-    end.
-
 wait_for_settlement(Tag) ->
-    wait_for_settlement(Tag, accepted).
-
-wait_for_settlement(Tag, State) ->
     receive
-        {amqp10_disposition, {State, Tag}} ->
+        {amqp10_disposition, {accepted, Tag}} ->
             flush(?FUNCTION_NAME),
             ok
     after 5000 ->
               flush("wait_for_settlement timed out"),
-              ct:fail(settled_timeout)
+              ct:fail(credited_timeout)
     end.
 
 wait_for_accepts(0) -> ok;