Merge pull request #4254 from rabbitmq/mqtt_quorum

Add quorum queues support for MQTT

Author: michaelklishin

deps/rabbitmq_mqtt/priv/schema/rabbitmq_mqtt.schema

@@ -257,3 +257,9 @@ fun(Conf) ->
     LingerTimeout = cuttlefish:conf_get("mqtt.tcp_listen_options.linger.timeout", Conf, 0),
     {LingerOn, LingerTimeout}
 end}.
+
+%% Durable queue type to be used for QoS 1 subscriptions
+%%
+
+{mapping, "mqtt.durable_queue_type", "rabbitmq_mqtt.durable_queue_type",
+    [{datatype, {enum, [classic, quorum]}}]}.

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -14,7 +14,7 @@
 
 %% for testing purposes
 -export([get_vhost_username/1, get_vhost/3, get_vhost_from_user_mapping/2,
-         add_client_id_to_adapter_info/2]).
+         add_client_id_to_adapter_info/2, maybe_quorum/3]).
 
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include("rabbit_mqtt_frame.hrl").
@@ -399,7 +399,7 @@ amqp_callback({#'basic.deliver'{ consumer_tag = ConsumerTag,
                            message_id    = MsgId,
                            send_fun      = SendFun,
                            amqp2mqtt_fun = Amqp2MqttFun } = PState) ->
-    amqp_channel:notify_received(DeliveryCtx),
+    notify_received(DeliveryCtx),
     case {delivery_dup(Delivery), delivery_qos(ConsumerTag, Headers, PState)} of
         {true, {?QOS_0, ?QOS_1}} ->
             amqp_channel:cast(
@@ -784,6 +784,22 @@ delivery_mode(?QOS_0) -> 1;
 delivery_mode(?QOS_1) -> 2;
 delivery_mode(?QOS_2) -> 2.
 
+maybe_quorum(Qos1Args, CleanSession, Queue) ->
+    case {rabbit_mqtt_util:env(durable_queue_type), CleanSession} of
+      %% it is possible to Quorum queues only if Clean Session == False
+      %% else always use Classic queues
+      %% Clean Session == True sets auto-delete to True and quorum queues
+      %% does not support auto-delete flag
+       {quorum, false} -> lists:append(Qos1Args,
+          [{<<"x-queue-type">>, longstr, <<"quorum">>}]);
+
+      {quorum, true} ->
+          rabbit_log:debug("Can't use quorum queue for ~s. " ++
+          "The clean session is true. Classic queue will be used", [Queue]),
+          Qos1Args;
+      _ -> Qos1Args
+    end.
+
 %% different qos subscriptions are received in different queues
 %% with appropriate durability and timeout arguments
 %% this will lead to duplicate messages for overlapping subscriptions
@@ -819,7 +835,7 @@ ensure_queue(Qos, #proc_state{ channels      = {Channel, _},
                                    %%
                                    %% see rabbitmq/rabbitmq-mqtt#37
                                    auto_delete = CleanSess,
-                                   arguments   = Qos1Args },
+                                   arguments   = maybe_quorum(Qos1Args, CleanSess, QueueQ1)},
                  #'basic.consume'{ queue  = QueueQ1,
                                    no_ack = false }};
             {_, _, ?QOS_0} ->
@@ -1091,3 +1107,10 @@ additional_info(Key,
                 #proc_state{adapter_info =
                             #amqp_adapter_info{additional_info = AddInfo}}) ->
     proplists:get_value(Key, AddInfo).
+
+notify_received(undefined) ->
+    %% no notification for quorum queues and streams
+    ok;
+notify_received(DeliveryCtx) ->
+    %% notification for flow control
+    amqp_channel:notify_received(DeliveryCtx).

deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl

@@ -123,6 +123,11 @@ handle_cast({close_connection, Reason},
 handle_cast(Msg, State) ->
     {stop, {mqtt_unexpected_cast, Msg}, State}.
 
+handle_info({#'basic.deliver'{}, #amqp_msg{}} = Delivery,
+    State) ->
+    %% receiving a message from a quorum queue
+    %% no delivery context
+    handle_info(erlang:insert_element(3, Delivery, undefined), State);
 handle_info({#'basic.deliver'{}, #amqp_msg{}, _DeliveryCtx} = Delivery,
             State = #state{ proc_state = ProcState }) ->
     callback_reply(State, rabbit_mqtt_processor:amqp_callback(Delivery,

deps/rabbitmq_mqtt/test/config_schema_SUITE_data/rabbitmq_mqtt.snippets

@@ -14,6 +14,7 @@
    mqtt.listeners.tcp.default = 1883
    mqtt.tcp_listen_options.backlog = 128
    mqtt.tcp_listen_options.nodelay = true
+   mqtt.durable_queue_type = classic
    mqtt.proxy_protocol = false",
   [{rabbit,[{tcp_listeners,[5672]}]},
    {rabbitmq_mqtt,
@@ -28,6 +29,7 @@
         {ssl_listeners,[]},
         {tcp_listeners,[1883]},
         {tcp_listen_options,[{backlog,128},{nodelay,true}]},
+        {durable_queue_type,classic},
         {proxy_protocol,false}]}],
   [rabbitmq_mqtt]},
 

deps/rabbitmq_mqtt/test/processor_SUITE.erl

@@ -24,7 +24,8 @@ groups() ->
                                 interprets_colons_in_username_if_option_not_set,
                                 get_vhosts_from_global_runtime_parameter,
                                 get_vhost,
-                                add_client_id_to_adapter_info
+                                add_client_id_to_adapter_info,
+                                quorum_configuration
                                ]}
     ].
 
@@ -209,3 +210,22 @@ clear_vhost_global_parameters() ->
         ok = mnesia:delete(rabbit_runtime_parameters, mqtt_port_to_vhost_mapping, write)
                          end,
     {atomic, ok} = mnesia:transaction(DeleteParameterFun).
+
+quorum_configuration(_Config) ->
+    MyArgs = [],
+%%  default setting with CleanSession = true of false
+    QMustBeClassic = rabbit_mqtt_processor:maybe_quorum(MyArgs, true, <<"">>),
+    ?assertEqual(QMustBeClassic, []),
+%%  default setting with CleanSession = true of false
+    QMustBeClassicEvenFalse = rabbit_mqtt_processor:maybe_quorum(MyArgs, false, <<"">>),
+    ?assertEqual(QMustBeClassicEvenFalse, []),
+    application:set_env(rabbitmq_mqtt, durable_queue_type, quorum),
+%%  quorum setting with CleanSession  == false must me quorum
+    QMustBeQuorum = rabbit_mqtt_processor:maybe_quorum(MyArgs, false, <<"">>),
+    ?assertEqual(QMustBeQuorum, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+
+    %%  quorum setting with CleanSession  == true must me classic since
+    %% quorum does not support auto-delete
+    QEvenQuorumMustBeClassic = rabbit_mqtt_processor:maybe_quorum(MyArgs, true, <<"">>),
+    ?assertEqual(QEvenQuorumMustBeClassic, []),
+    ok.

deps/rabbitmq_mqtt/test/reader_SUITE.erl

@@ -12,7 +12,8 @@
 
 all() ->
     [
-      {group, non_parallel_tests}
+      {group, non_parallel_tests},
+      {group, non_parallel_tests_quorum}
     ].
 
 groups() ->
@@ -21,7 +22,13 @@ groups() ->
                                 block,
                                 handle_invalid_frames,
                                 stats
-                               ]}
+                               ]},
+      {non_parallel_tests_quorum, [], [
+                                quorum_session_false,
+                                quorum_session_true,
+                                classic_session_true,
+                                classic_session_false
+      ]}
     ].
 
 suite() ->
@@ -55,6 +62,14 @@ end_per_suite(Config) ->
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_group(non_parallel_tests_quorum, Config) ->
+%%  Added for quorum queue test else the mixing test would fail
+%% with "feature flag is disabled"
+    case rabbit_ct_broker_helpers:enable_feature_flag(Config, quorum_queue) of
+        ok -> Config;
+        Skip -> Skip
+    end,
+  Config;
 init_per_group(_, Config) ->
     Config.
 
@@ -154,6 +169,64 @@ stats(Config) ->
                               [connection_coarse_metrics, Pid]),
     emqttc:disconnect(C).
 
+get_durable_queue_type(Server, Q0) ->
+    QNameRes = rabbit_misc:r(<<"/">>, queue, Q0),
+    {ok, Q1} = rpc:call(Server, rabbit_amqqueue, lookup, [QNameRes]),
+    amqqueue:get_type(Q1).
+
+set_env(QueueType) ->
+    application:set_env(rabbitmq_mqtt, durable_queue_type, QueueType).
+
+get_env() ->
+    rabbit_mqtt_util:env(durable_queue_type).
+
+
+validate_durable_queue_type(Config, ClientName, CleanSession, Expected) ->
+    P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    {ok, C} = emqttc:start_link([{host, "localhost"},
+    {port, P},
+    {clean_sess, CleanSession},
+    {client_id, ClientName},
+    {proto_ver, 3},
+    {logger, info},
+    {puback_timeout, 1}]),
+    emqttc:subscribe(C, <<"TopicB">>, qos1),
+    emqttc:publish(C, <<"TopicB">>, <<"Payload">>),
+    expect_publishes(<<"TopicB">>, [<<"Payload">>]),
+    emqttc:unsubscribe(C, [<<"TopicB">>]),
+    Prefix = <<"mqtt-subscription-">>,
+    Suffix = <<"qos1">>,
+    Q= <<Prefix/binary, ClientName/binary, Suffix/binary>>,
+    ?assertEqual(Expected,get_durable_queue_type(Server,Q)),
+    timer:sleep(500),
+    emqttc:disconnect(C).
+
+%% quorum queue test when enable
+quorum_session_false(Config) ->
+  %%  test if the quorum queue is enable after the setting
+    Default = rpc(Config, reader_SUITE, get_env, []),
+    rpc(Config, reader_SUITE, set_env, [quorum]),
+    validate_durable_queue_type(Config, <<"qCleanSessionFalse">>, false, rabbit_quorum_queue),
+    rpc(Config, reader_SUITE, set_env, [Default]).
+
+quorum_session_true(Config) ->
+  %%  in case clean session == true must be classic since quorum
+  %% doesn't support auto-delete
+    Default = rpc(Config, reader_SUITE, get_env, []),
+    rpc(Config, reader_SUITE, set_env, [quorum]),
+    validate_durable_queue_type(Config, <<"qCleanSessionTrue">>, true, rabbit_classic_queue),
+    rpc(Config, reader_SUITE, set_env, [Default]).
+
+classic_session_true(Config) ->
+  %%  with default configuration the queue is classic
+    validate_durable_queue_type(Config, <<"cCleanSessionTrue">>, true, rabbit_classic_queue).
+
+classic_session_false(Config) ->
+  %%  with default configuration the queue is classic
+    validate_durable_queue_type(Config, <<"cCleanSessionFalse">>, false, rabbit_classic_queue).
+
+
 expect_publishes(_Topic, []) -> ok;
 expect_publishes(Topic, [Payload|Rest]) ->
     receive