QQ/Streams: Ensure open file handles are closed when a queue is deleted.

Whilst a session or channel has one open.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo_client.erl

@@ -14,6 +14,7 @@
 -export([
          init/1,
          init/2,
+         close/1,
          checkout/4,
          cancel_checkout/3,
          enqueue/3,
@@ -755,6 +756,13 @@ handle_ra_event(QName, Leader, close_cached_segments,
 handle_ra_event(_QName, _Leader, {machine, eol}, State) ->
     {eol, [{unblock, cluster_name(State)}]}.
 
+-spec close(rabbit_fifo_client:state()) -> ok.
+close(#state{cached_segments = undefined}) ->
+    ok;
+close(#state{cached_segments = {_, _, Flru}}) ->
+    _ = ra_flru:evict_all(Flru),
+    ok.
+
 %% @doc Attempts to enqueue a message using cast semantics. This provides no
 %% guarantees or retries if the message fails to achieve consensus or if the
 %% servers sent to happens not to be available. If the message is sent to a

deps/rabbit/src/rabbit_queue_type.erl

@@ -407,7 +407,9 @@ remove(QRef, #?STATE{ctxs = Ctxs0} = State) ->
     case maps:take(QRef, Ctxs0) of
         error ->
             State;
-        {_, Ctxs} ->
+        {#ctx{module = Mod,
+              state = S}, Ctxs} ->
+            ok = Mod:close(S),
             State#?STATE{ctxs = Ctxs}
     end.
 
@@ -495,11 +497,10 @@ init() ->
 
 -spec close(state()) -> ok.
 close(#?STATE{ctxs = Contexts}) ->
-    maps:foreach(
-      fun (_, #ctx{module = Mod,
-                   state = S}) ->
-              ok = Mod:close(S)
-      end, Contexts).
+    maps:foreach(fun (_, #ctx{module = Mod,
+                              state = S}) ->
+                         ok = Mod:close(S)
+                 end, Contexts).
 
 -spec new(amqqueue:amqqueue(), state()) -> state().
 new(Q, State) when ?is_amqqueue(Q) ->

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -159,7 +159,6 @@
 -define(RPC_TIMEOUT, 1000).
 -define(START_CLUSTER_TIMEOUT, 5000).
 -define(START_CLUSTER_RPC_TIMEOUT, 60_000). %% needs to be longer than START_CLUSTER_TIMEOUT
--define(FORCE_CHECKPOINT_RPC_TIMEOUT, 15_000).
 -define(TICK_INTERVAL, 5000). %% the ra server tick time
 -define(DELETE_TIMEOUT, 5000).
 -define(MEMBER_CHANGE_TIMEOUT, 20_000).
@@ -230,8 +229,8 @@ init(Q) when ?is_amqqueue(Q) ->
     {ok, rabbit_fifo_client:init(Servers, SoftLimit)}.
 
 -spec close(rabbit_fifo_client:state()) -> ok.
-close(_State) ->
-    ok.
+close(State) ->
+    rabbit_fifo_client:close(State).
 
 -spec update(amqqueue:amqqueue(), rabbit_fifo_client:state()) ->
     rabbit_fifo_client:state().