Fix memory leak when using mandatory flag in CQs

MsgNo is defined for both mandatory flags and confirms,
but only confirms need to be kept track of in this module.

(cherry picked from commit 7ffb154787446522e3838ef2230226b337665e4d)

2 files changed, 43 insertions, 7 deletions

diff --git a/deps/rabbit/src/rabbit_classic_queue.erl b/deps/rabbit/src/rabbit_classic_queue.erl
index abf198b692..3a4092745a 100644
--- a/deps/rabbit/src/rabbit_classic_queue.erl
+++ b/deps/rabbit/src/rabbit_classic_queue.erl
@@ -299,9 +299,9 @@ settlement_action(Type, QRef, MsgSeqs, Acc) ->
 deliver(Qs0, #delivery{flow = Flow,
                        msg_seq_no = MsgNo,
                        message = #basic_message{exchange_name = _Ex},
-                       confirm = _Confirm} = Delivery) ->
+                       confirm = Confirm} = Delivery) ->
     %% TODO: record master and slaves for confirm processing
-    {MPids, SPids, Qs, Actions} = qpids(Qs0, MsgNo),
+    {MPids, SPids, Qs, Actions} = qpids(Qs0, Confirm, MsgNo),
     QPids = MPids ++ SPids,
     case Flow of
         %% Here we are tracking messages sent by the rabbit_channel
@@ -361,7 +361,7 @@ purge(Q) when ?is_amqqueue(Q) ->
     QPid = amqqueue:get_pid(Q),
     delegate:invoke(QPid, {gen_server2, call, [purge, infinity]}).
 
-qpids(Qs, MsgNo) ->
+qpids(Qs, Confirm, MsgNo) ->
     lists:foldl(
       fun ({Q, S0}, {MPidAcc, SPidAcc, Qs0, Actions0}) ->
               QPid = amqqueue:get_pid(Q),
@@ -369,14 +369,14 @@ qpids(Qs, MsgNo) ->
               QRef = amqqueue:get_name(Q),
               Actions = [{monitor, QPid, QRef}
                          | [{monitor, P, QRef} || P <- SPids]] ++ Actions0,
-              %% confirm record only if MsgNo isn't undefined
+              %% confirm record only if necessary
               S = case S0 of
                       #?STATE{unconfirmed = U0} ->
                           Rec = [QPid | SPids],
-                          U = case MsgNo of
-                                  undefined ->
+                          U = case Confirm of
+                                  false ->
                                       U0;
-                                  _ ->
+                                  true ->
                                       U0#{MsgNo => #msg_status{pending = Rec}}
                               end,
                           S0#?STATE{pid = QPid,
diff --git a/release-notes/3.9.8.md b/release-notes/3.9.8.md
new file mode 100644
index 0000000000..31cba1e0bb
--- /dev/null
+++ b/release-notes/3.9.8.md
@@ -0,0 +1,36 @@
+RabbitMQ `3.9.8` is a maintenance release in the `3.9.x` release series.
+
+Please refer to the **Upgrading to 3.9** section from [v3.9.0 release notes](https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.9.0) if upgrading from a version prior to 3.9.0.
+
+This release requires at least Erlang 23.2, and supports the latest Erlang 24 version, 24.1.2 at the time of release. [RabbitMQ and Erlang/OTP Compatibility Matrix](https://www.rabbitmq.com/which-erlang.html) has more details on Erlang version requirements for RabbitMQ.
+
+
+
+## Changes Worth Mentioning
+
+Release notes are kept under [rabbitmq-server/release-notes](https://github.com/rabbitmq/rabbitmq-server/tree/v3.9.x/release-notes).
+Contributors are encouraged to update them together with their changes. This helps with release automation and a more consistent release schedule.
+
+### Core Server
+
+#### Enhancements
+
+ * ...
+
+
+#### Bug Fixes
+
+* When the mandatory flag was used when publishing to classic queues,
+  but publisher confirms were not, channels memory usage would grow indefinitely.
+
+   GitHub issue: [#3560](https://github.com/rabbitmq/rabbitmq-server/issues/3560)
+
+
+## Dependency Upgrades
+
+ * ...
+
+
+## Source Code Archives
+
+To obtain source code of the entire distribution, please download the archive named `rabbitmq-server-3.9.8.tar.xz` instead of the source tarball produced by GitHub.