QQ: use a dedicated function for queue recovery after Ra system restart.

Previously we used the `registered` approach where all Ra servers that
have a registered name would be recovered. This could have unintended
side effects for queues that e.g. were deleted when not all members of
a quorum queueu were running when the queue was deleted. In this case
the Ra system would have recovered the members that were not deleted
which is not ideal as a dangling member would just sit and loop in
pre vote state and a future declaration of the queue may partially
fail.

Instead we rely on the meta data store for the truth about which
members should be restarted after a ra system restart.

Author: kjnilsson

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -17,6 +17,7 @@
          handle_event/3]).
 -export([is_recoverable/1,
          recover/2,
+         system_recover/1,
          stop/1,
          start_server/1,
          restart_server/1,
@@ -97,6 +98,14 @@
 -define(RA_SYSTEM, quorum_queues).
 -define(RA_WAL_NAME, ra_log_wal).
 
+-define(INFO(Str, Args),
+        rabbit_log:info("[~s:~s/~b] " Str,
+                        [?MODULE, ?FUNCTION_NAME, ?FUNCTION_ARITY | Args])).
+-define(WARN(Str, Args),
+        rabbit_log:warning("[~s:~s/~b] " Str,
+                           [?MODULE, ?FUNCTION_NAME, ?FUNCTION_ARITY | Args])).
+
+
 -define(STATISTICS_KEYS,
         [policy,
          operator_policy,
@@ -641,6 +650,21 @@ is_recoverable(Q) when ?is_amqqueue(Q) and ?amqqueue_is_quorum(Q) ->
     Nodes = get_nodes(Q),
     lists:member(Node, Nodes).
 
+system_recover(quorum_queues) ->
+    case rabbit:is_booted() of
+        true ->
+            Queues = rabbit_amqqueue:list_local_quorum_queues(),
+            ?INFO("recovering ~b queues", [length(Queues)]),
+            {Recovered, Failed} = recover(<<>>, Queues),
+            ?INFO("recovered ~b queues, "
+                  "failed to recover ~b queues",
+                  [length(Recovered), length(Failed)]),
+            ok;
+        false ->
+            ?INFO("rabbit not booted, skipping queue recovery", []),
+            ok
+    end.
+
 -spec recover(binary(), [amqqueue:amqqueue()]) ->
     {[amqqueue:amqqueue()], [amqqueue:amqqueue()]}.
 recover(_Vhost, Queues) ->

deps/rabbit/src/rabbit_ra_systems.erl

@@ -130,7 +130,8 @@ get_config(quorum_queues = RaSystem) ->
                    wal_max_entries => WalMaxEntries,
                    segment_compute_checksums => SegmentChecksums,
                    compress_mem_tables => CompressMemTables,
-                   server_recovery_strategy => registered};
+                   server_recovery_strategy => {rabbit_quorum_queue,
+                                                system_recover, []}};
 get_config(coordination = RaSystem) ->
     DefaultConfig = get_default_config(),
     CoordDataDir = filename:join(