Fix classic queue declaration crash

[ansd]

Repro:

1. Deploy on kind:

--- apiVersion: rabbitmq.com/v1beta1 kind: RabbitmqCluster metadata: name: r1 spec: replicas: 3 image: rabbitmq:3.12.0-management rabbitmq: additionalPlugins: - rabbitmq_mqtt 

2. kubectl port-forward svc/r1 1883
3. kubectl exec r1-server-2 -c rabbitmq -- rabbitmqctl stop_app
4. mqttx sub -V 3.1.1 -t a/b -q 0 -i client1 -u <user> -P <password> --no-clean

resulted in the following crash (logs on r1-server-1):

2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> Server startup complete; 7 plugins started. 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_prometheus 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_peer_discovery_k8s 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_mqtt 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_peer_discovery_common 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_management 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_web_dispatch 2023-06-20 14:41:40.919043+00:00 [info] <0.632.0> * rabbitmq_management_agent 2023-06-20 14:41:42.518325+00:00 [info] <0.530.0> rabbit on node 'rabbit@r1-server-2.r1-nodes.default' up 2023-06-20 14:41:48.094816+00:00 [info] <0.530.0> rabbit on node 'rabbit@r1-server-0.r1-nodes.default' up 2023-06-20 14:42:36.851565+00:00 [info] <0.530.0> rabbit on node 'rabbit@r1-server-2.r1-nodes.default' down 2023-06-20 14:42:47.703047+00:00 [info] <0.1257.0> Accepted MQTT connection 127.0.0.1:42906 -> 127.0.0.1:1883 for client ID client2 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ** Generic server <0.1257.0> terminating 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ** Last message in was {tcp,#Port<0.71>,<<130,8,78,151,0,3,97,47,98,0>>} 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ** When Server state == #{await_recv => true, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> conn_name => <<"127.0.0.1:42906 -> 127.0.0.1:1883">>, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> connection_state => running,conserve => false, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> deferred_recv => false, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> keepalive => 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {state,60,#Ref<0.3356128136.2053636098.25604>, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> #Port<0.71>,89,true}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> proc_state => 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> #{auth_state => 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {auth_state, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {user,<<"default_user_8tMWbW0B9lOFNUgpc6j">>, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [administrator], 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{rabbit_auth_backend_internal, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> #Fun<rabbit_auth_backend_internal.3.114557357>}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> #{<<"client_id">> => <<"client2">>}}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> cfg => 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> #{clean_sess => false, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> client_id => <<"client2">>, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> conn_name => undefined, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> connected_at => 1687272167702, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> delivery_flow => flow, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> exchange => 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {resource,<<"/">>,exchange,<<"amq.topic">>}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ip_addr => {0,0,0,0,0,65535,32512,1}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> peer_ip_addr => {0,0,0,0,0,65535,32512,1}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> peer_port => 42906,port => 1883, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> prefetch => 10,proto_ver => mqtt311, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> published => false,retainer_pid => <0.742.0>, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> socket => #Port<0.71>,ssl_login_name => none, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> trace_state => none,will_msg_defined => false}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> packet_id => 1,qos0_messages_dropped => 0, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> queue_states => #{num_queue_clients => 0}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> queues_soft_limit_exceeded => #{}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ra_register_state => undefined, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> subscriptions => #{},unacked_client_pubs => #{}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> unacked_server_pubs => #{}}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> proxy_socket => undefined,socket => #Port<0.71>, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> stats_timer => 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {state,fine,5000, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> #Ref<0.3356128136.2053636098.25603>}} 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ** Reason for termination == 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> ** {{badmatch,'rabbit@r1-server-1.r1-nodes.default'}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{rabbit_classic_queue,declare,2, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{file,"rabbit_classic_queue.erl"},{line,83}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {rabbit_mqtt_processor,create_queue,2, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{file,"rabbit_mqtt_processor.erl"},{line,922}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {rabbit_mqtt_processor,'-process_request/3-fun-0-',2, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{file,"rabbit_mqtt_processor.erl"},{line,305}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {lists,foldl,3,[{file,"lists.erl"},{line,1350}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {rabbit_mqtt_processor,process_request,3, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{file,"rabbit_mqtt_processor.erl"},{line,293}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {rabbit_mqtt_reader,process_received_bytes,2, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> [{file,"rabbit_mqtt_reader.erl"},{line,353}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {gen_server,try_dispatch,4,[{file,"gen_server.erl"},{line,1123}]}, 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> {gen_server,handle_msg,6,[{file,"gen_server.erl"},{line,1200}]}]} 2023-06-20 14:42:47.705161+00:00 [error] <0.1257.0> 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> crasher: 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> initial call: rabbit_mqtt_reader:init/1 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> pid: <0.1257.0> 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> registered_name: [] 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> exception error: no match of right hand side value 'rabbit@r1-server-1.r1-nodes.default' 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in function rabbit_classic_queue:declare/2 (rabbit_classic_queue.erl, line 83) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from rabbit_mqtt_processor:create_queue/2 (rabbit_mqtt_processor.erl, line 922) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from rabbit_mqtt_processor:'-process_request/3-fun-0-'/2 (rabbit_mqtt_processor.erl, line 305) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from lists:foldl/3 (lists.erl, line 1350) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from rabbit_mqtt_processor:process_request/3 (rabbit_mqtt_processor.erl, line 293) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from rabbit_mqtt_reader:process_received_bytes/2 (rabbit_mqtt_reader.erl, line 353) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from gen_server:try_dispatch/4 (gen_server.erl, line 1123) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> in call from gen_server:handle_msg/6 (gen_server.erl, line 1200) 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> ancestors: [<0.750.0>,<0.749.0>,<0.748.0>,<0.746.0>,<0.745.0>, 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> rabbit_mqtt_sup,<0.738.0>] 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> message_queue_len: 0 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> messages: [] 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> links: [<0.750.0>,#Port<0.71>] 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> dictionary: [{'$logger_metadata$', 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> #{domain => [rabbitmq,connection,mqtt]}}, 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> {permission_cache, 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> [{{resource,<<"/">>,queue, 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> <<"mqtt-subscription-client2qos0">>}, 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> #{<<"client_id">> => <<"client2">>}, 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> configure}]}] 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> trap_exit: true 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> status: running 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> heap_size: 17731 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> stack_size: 28 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> reductions: 61261 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> neighbours: 2023-06-20 14:42:47.705725+00:00 [error] <0.1257.0> 2023-06-20 14:42:47.706116+00:00 [error] <0.750.0> Ranch listener {acceptor,{0,0,0,0,0,0,0,0},1883} had connection process started with rabbit_mqtt_reader:start_link/3 at <0.1257.0> exit with reason: {{badmatch,'rabbit@r1-server-1.r1-nodes.default'},[{rabbit_classic_queue,declare,2,[{file,"rabbit_classic_queue.erl"},{line,83}]},{rabbit_mqtt_processor,create_queue,2,[{file,"rabbit_mqtt_processor.erl"},{line,922}]},{rabbit_mqtt_processor,'-process_request/3-fun-0-',2,[{file,"rabbit_mqtt_processor.erl"},{line,305}]},{lists,foldl,3,[{file,"lists.erl"},{line,1350}]},{rabbit_mqtt_processor,process_request,3,[{file,"rabbit_mqtt_processor.erl"},{line,293}]},{rabbit_mqtt_reader,process_received_bytes,2,[{file,"rabbit_mqtt_reader.erl"},{line,353}]},{gen_server,try_dispatch,4,[{file,"gen_server.erl"},{line,1123}]},{gen_server,handle_msg,6,[{file,"gen_server.erl"},{line,1200}]}]} 2023-06-20 14:42:47.706116+00:00 [error] <0.750.0> 

A similar CT test always succeeded in deps/rabbitmq_mqtt/test/shared_SUITE.erl, that's why I left it out:

non_clean_sess_connect_qos1(Config) -> ClientId = ?FUNCTION_NAME, ok = rabbit_ct_broker_helpers:stop_broker(Config, 2), C1 = connect(ClientId, Config, [{clean_start, false}]), {ok, _, [1]} = emqtt:subscribe(C1, [{<<"a/b">>, qos1}]), ok = rabbit_ct_broker_helpers:start_broker(Config, 2), ok = emqtt:disconnect(C1), C2 = connect(ClientId, Config, [{clean_start, true}]), ok = emqtt:disconnect(C2).