consumewr timeouts mqtt

Author: kjnilsson

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -2026,20 +2026,30 @@ handle_queue_event({queue_event, QName, Evt},
     case rabbit_queue_type:handle_event(QName, Evt, QStates0) of
         {ok, QStates, Actions} ->
             State1 = State0#state{queue_states = QStates},
-            State = handle_queue_actions(Actions, State1),
-            {ok, State};
+            try handle_queue_actions(Actions, State1) of
+                State ->
+                    {ok, State}
+            catch throw:Reason when Reason =:= consuming_queue_down;
+                                    Reason =:= consumer_timeout ->
+                    {error, Reason, State1}
+            end;
         {eol, Actions} ->
-            State1 = handle_queue_actions(Actions, State0),
-            {ConfirmPktIds, U} = rabbit_mqtt_confirms:remove_queue(QName, U0),
-            QStates = rabbit_queue_type:remove(QName, QStates0),
-            State = State1#state{queue_states = QStates,
-                                 unacked_client_pubs = U},
-            send_puback(ConfirmPktIds, ?RC_SUCCESS, State),
-            try handle_queue_down(QName, State) of
-                State2 ->
-                    {ok, State2}
-            catch throw:consuming_queue_down ->
-                    {error, consuming_queue_down, State}
+            try handle_queue_actions(Actions, State0) of
+                State1 ->
+                    {ConfirmPktIds, U} = rabbit_mqtt_confirms:remove_queue(QName, U0),
+                    QStates = rabbit_queue_type:remove(QName, QStates0),
+                    State = State1#state{queue_states = QStates,
+                                         unacked_client_pubs = U},
+                    send_puback(ConfirmPktIds, ?RC_SUCCESS, State),
+                    try handle_queue_down(QName, State) of
+                        State2 ->
+                            {ok, State2}
+                    catch throw:consuming_queue_down ->
+                            {error, consuming_queue_down, State}
+                    end
+            catch throw:Reason when Reason =:= consuming_queue_down;
+                                    Reason =:= consumer_timeout ->
+                    {error, Reason, State0}
             end;
         {protocol_error, _Type, _Reason, _ReasonArgs} = Error ->
             {error, Error, State0}
@@ -2071,7 +2081,11 @@ handle_queue_actions(Actions, #state{} = State0) ->
           ({unblock, QName}, S = #state{queues_soft_limit_exceeded = QSLE}) ->
               S#state{queues_soft_limit_exceeded = sets:del_element(QName, QSLE)};
           ({queue_down, QName}, S) ->
-              handle_queue_down(QName, S)
+              handle_queue_down(QName, S);
+          ({released, QName, _CTag, _MsgSeqNos, timeout}, _S) ->
+              ?LOG_INFO("Terminating MQTT connection because consumer on ~s timed out",
+                        [rabbit_misc:rs(QName)]),
+              throw(consumer_timeout)
       end, State0, Actions).
 
 handle_queue_down(QName, State0 = #state{cfg = #cfg{client_id = ClientId}}) ->

deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl

@@ -131,7 +131,8 @@ handle_cast(QueueEvent = {queue_event, _, _},
     try rabbit_mqtt_processor:handle_queue_event(QueueEvent, PState0) of
         {ok, PState} ->
             maybe_process_deferred_recv(control_throttle(pstate(State, PState)));
-        {error, consuming_queue_down = Reason, PState} ->
+        {error, Reason, PState} when Reason =:= consuming_queue_down;
+                                     Reason =:= consumer_timeout ->
             {stop, {shutdown, Reason}, pstate(State, PState)};
         {error, Reason0, PState} ->
             {stop, Reason0, pstate(State, PState)}

deps/rabbitmq_mqtt/test/v5_SUITE.erl

@@ -126,7 +126,8 @@ cluster_size_1_tests() ->
      topic_alias_disallowed_retained_message,
      extended_auth,
      headers_exchange,
-     consistent_hash_exchange
+     consistent_hash_exchange,
+     consumer_timeout_quorum_queue
     ].
 
 cluster_size_3_tests() ->
@@ -2158,6 +2159,58 @@ consistent_hash_exchange(Config) ->
     [#'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = Q}) || Q <- Qs],
     ok = rabbit_ct_client_helpers:close_channels_and_connection(Config, 0).
 
+%% Test that consumer timeout on a quorum queue terminates the MQTT connection.
+%% Consumer timeouts are only supported for quorum queues.
+%%
+%% We use {auto_ack, never} to prevent emqtt from automatically sending PUBACK,
+%% which allows the consumer timeout to trigger.
+consumer_timeout_quorum_queue(Config) ->
+    Topic = atom_to_binary(?FUNCTION_NAME),
+    PolicyName = <<"consumer-timeout-policy">>,
+
+    ok = rpc(Config, application, set_env, [?APP, durable_queue_type, quorum]),
+    %% Set a short consumer timeout policy for quorum queues
+    ok = rabbit_ct_broker_helpers:set_policy(
+           Config, 0, PolicyName, <<".*">>, <<"quorum_queues">>,
+           [{<<"consumer-timeout">>, 1000}]),
+
+    %% Use two clients: one for publishing (normal auto_ack), one for subscribing (auto_ack=never)
+    Pub = connect(<<"publisher">>, Config),
+    Sub = connect(?FUNCTION_NAME, Config,
+                  non_clean_sess_opts() ++ [{auto_ack, never}]),
+    {ok, _, [1]} = emqtt:subscribe(Sub, Topic, qos1),
+
+    %% Publish a message from the publisher client
+    {ok, _} = emqtt:publish(Pub, Topic, <<"test message">>, [{qos, 1}]),
+
+    %% Receive the message on subscriber but do NOT acknowledge it (auto_ack=never)
+    receive
+        {publish, #{client_pid := Sub,
+                    topic := Topic,
+                    payload := <<"test message">>}} ->
+            ok
+    after 5000 ->
+              ct:fail("did not receive message")
+    end,
+
+    ok = rpc(Config, application, unset_env, [?APP, durable_queue_type]),
+
+    %% Trap exits so we receive EXIT message instead of crashing when Sub exits
+    process_flag(trap_exit, true),
+
+    %% The subscriber connection should be terminated due to consumer timeout
+    %% Wait for the consumer timeout (1s) plus some margin
+    util:await_exit(Sub),
+
+    process_flag(trap_exit, false),
+
+    %% Cleanup
+    ok = emqtt:disconnect(Pub),
+    ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, PolicyName),
+    %% Clean up the session by connecting with clean_start
+    C2 = connect(?FUNCTION_NAME, Config, [{clean_start, true}]),
+    ok = emqtt:disconnect(C2).
+
 %% -------------------------------------------------------------------
 %% Helpers
 %% -------------------------------------------------------------------