Remove RabbitMQ internal flow control for classic queue

Remove RabbitMQ internal flow control between AMQP writer and classic
queue for credit API v1

Author: ansd

deps/rabbit/src/rabbit_amqp_session.erl

@@ -1829,38 +1829,18 @@ echo(Echo, HandleInt, DeliveryCount, LinkCredit, Available, State) ->
 
 send_pending_delivery(#pending_delivery{
                          frames = Frames,
-                         queue_pid = QPid,
-                         outgoing_unsettled = #outgoing_unsettled{consumer_tag = Ctag,
-                                                                  queue_name = QName}
+                         outgoing_unsettled = #outgoing_unsettled{consumer_tag = Ctag}
                         } = Pending,
                       Buf0,
                       #state{remote_incoming_window = Space,
                              outgoing_links = OutgoingLinks,
-                             queue_states = QStates,
                              cfg = #cfg{writer_pid = WriterPid,
                                         channel_num = Ch}
                             } = State0) ->
     Handle = ctag_to_handle(Ctag),
     case is_map_key(Handle, OutgoingLinks) of
         true ->
-            SendFun = case QPid of
-                          credit_api_v2 ->
-                              send_fun(WriterPid, Ch);
-                          _ ->
-                              case rabbit_queue_type:module(QName, QStates) of
-                                  {ok, rabbit_classic_queue} ->
-                                      %% Classic queue client and classic queue process that
-                                      %% communicate via credit API v1 use RabbitMQ internal
-                                      %% credit flow control.
-                                      fun(Transfer, Sections) ->
-                                              rabbit_amqp_writer:send_command_and_notify(
-                                                WriterPid, QPid, Ch, Transfer, Sections)
-                                      end;
-                                  {ok, _QType} ->
-                                      send_fun(WriterPid, Ch)
-                              end
-                      end,
-            case send_frames(SendFun, Frames, Space) of
+            case send_frames(send_fun(WriterPid, Ch), Frames, Space) of
                 {sent_all, SpaceLeft} ->
                     State1 = State0#state{outgoing_pending = Buf0},
                     State = sent_pending_delivery(Pending, Handle, State1),

deps/rabbit/src/rabbit_amqp_writer.erl

@@ -15,7 +15,6 @@
          send_command/3,
          send_command/4,
          send_command_sync/3,
-         send_command_and_notify/5,
          assemble_frame/3]).
 
 %% gen_server callbacks
@@ -75,16 +74,6 @@ send_command_sync(Writer, ChannelNum, Performative) ->
     Request = {send_command, ChannelNum, Performative},
     gen_server:call(Writer, Request, ?CALL_TIMEOUT).
 
-%% Delete this function when feature flag rabbitmq_4.0.0 becomes required.
--spec send_command_and_notify(pid(),
-                              pid(),
-                              rabbit_types:channel_number(),
-                              performative(),
-                              payload()) -> ok | {error, blocked}.
-send_command_and_notify(Writer, QueuePid, ChannelNum, Performative, Payload) ->
-    Request = {send_command_and_notify, QueuePid, self(), ChannelNum, Performative, Payload},
-    maybe_send(Writer, Request).
-
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %%% gen_server callbacks %%%
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%
@@ -105,12 +94,6 @@ handle_cast({send_command, ChannelNum, Performative}, State0) ->
 handle_cast({send_command, SessionPid, ChannelNum, Performative, Payload}, State0) ->
     State1 = internal_send_command_async(ChannelNum, Performative, Payload, State0),
     State = credit_flow_ack(SessionPid, State1),
-    no_reply(State);
-%% Delete below function clause when feature flag rabbitmq_4.0.0 becomes required.
-handle_cast({send_command_and_notify, QueuePid, SessionPid, ChannelNum, Performative, Payload}, State0) ->
-    State1 = internal_send_command_async(ChannelNum, Performative, Payload, State0),
-    State = credit_flow_ack(SessionPid, State1),
-    rabbit_amqqueue:notify_sent(QueuePid, SessionPid),
     no_reply(State).
 
 handle_call({send_command, ChannelNum, Performative}, _From, State0) ->
@@ -132,10 +115,6 @@ handle_info({{'DOWN', session}, _MRef, process, SessionPid, _Reason},
             State0 = #state{monitored_sessions = Sessions}) ->
     credit_flow:peer_down(SessionPid),
     State = State0#state{monitored_sessions = maps:remove(SessionPid, Sessions)},
-    no_reply(State);
-%% Delete below function clause when feature flag rabbitmq_4.0.0 becomes required.
-handle_info({'DOWN', _MRef, process, QueuePid, _Reason}, State) ->
-    rabbit_amqqueue:notify_sent_queue_down(QueuePid),
     no_reply(State).
 
 format_status(Status) ->

deps/rabbit/src/rabbit_ff_controller.erl

@@ -782,10 +782,10 @@ get_stable_feature_flags(#{feature_flags := FeatureFlags}) ->
 %% There are two ways to specify that list:
 %% <ol>
 %% <li>Using the `$RABBITMQ_FEATURE_FLAGS' environment variable; for
-%%   instance `RABBITMQ_FEATURE_FLAGS=rabbitmq_4.0.0,khepri_db'.</li>
+%%   instance `RABBITMQ_FEATURE_FLAGS=rabbitmq_4.2.0,khepri_db'.</li>
 %% <li>Using the `forced_feature_flags_on_init' configuration parameter;
 %%   for instance
-%%   `{rabbit, [{forced_feature_flags_on_init, [rabbitmq_4.0.0, khepri_db]}]}'.</li>
+%%   `{rabbit, [{forced_feature_flags_on_init, [rabbitmq_4.2.0, khepri_db]}]}'.</li>
 %% </ol>
 %%
 %% There's also a way to enable specific flags and skip others: