From 2c661910436650c6a66dcc4036ce46a55a265a9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-S=C3=A9bastien=20P=C3=A9dron?= Date: Wed, 26 Feb 2025 14:54:02 +0100 Subject: [PATCH 1/8] amqp_client_SUITE: Use a dedicated CI job for this testsuite [Why] This testsuite is very unstable and it is difficult to debug while it is part of a `parallel-ct` group. It also forced us to re-run the entire `parallel-ct` group just to retry that one testsuite. --- .github/workflows/test-make-tests.yaml | 1 + deps/rabbit/Makefile | 7 ++++--- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/.github/workflows/test-make-tests.yaml b/.github/workflows/test-make-tests.yaml index 5fa4c6e43d48..a4ffd93c453c 100644 --- a/.github/workflows/test-make-tests.yaml +++ b/.github/workflows/test-make-tests.yaml @@ -25,6 +25,7 @@ jobs: - parallel-ct-set-2 - parallel-ct-set-3 - parallel-ct-set-4 + - ct-amqp_client - ct-clustering_management - eunit ct-dead_lettering - ct-feature_flags diff --git a/deps/rabbit/Makefile b/deps/rabbit/Makefile index 5aebf56a99f5..828ce2fc6357 100644 --- a/deps/rabbit/Makefile +++ b/deps/rabbit/Makefile @@ -175,7 +175,8 @@ bats: $(BATS) tests:: bats -SLOW_CT_SUITES := backing_queue \ +SLOW_CT_SUITES := amqp_client \ + backing_queue \ channel_interceptor \ cluster \ cluster_rename \ @@ -257,7 +258,7 @@ define ct_master.erl halt(0) endef -PARALLEL_CT_SET_1_A = amqp_client unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking +PARALLEL_CT_SET_1_A = unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_credit_api_v2 amqp_filtex amqp_dotnet amqp_jms signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack amqpl_direct_reply_to backing_queue bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange rabbit_direct_reply_to_prop cluster_limit cluster_minority term_to_binary_compat_prop topic_permission transactions unicode unit_access_control PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit @@ -282,7 +283,7 @@ PARALLEL_CT_SET_2 = $(sort $(PARALLEL_CT_SET_2_A) $(PARALLEL_CT_SET_2_B) $(PARAL PARALLEL_CT_SET_3 = $(sort $(PARALLEL_CT_SET_3_A) $(PARALLEL_CT_SET_3_B) $(PARALLEL_CT_SET_3_C) $(PARALLEL_CT_SET_3_D)) PARALLEL_CT_SET_4 = $(sort $(PARALLEL_CT_SET_4_A) $(PARALLEL_CT_SET_4_B) $(PARALLEL_CT_SET_4_C) $(PARALLEL_CT_SET_4_D)) -SEQUENTIAL_CT_SUITES = clustering_management dead_lettering feature_flags metadata_store_clustering quorum_queue rabbit_stream_queue +SEQUENTIAL_CT_SUITES = amqp_client clustering_management dead_lettering feature_flags metadata_store_clustering quorum_queue rabbit_stream_queue PARALLEL_CT_SUITES = $(PARALLEL_CT_SET_1) $(PARALLEL_CT_SET_2) $(PARALLEL_CT_SET_3) $(PARALLEL_CT_SET_4) ifeq ($(filter-out $(SEQUENTIAL_CT_SUITES) $(PARALLEL_CT_SUITES),$(CT_SUITES)),) From 77e3636272b3932bbac0a41f727fc55539331c5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-S=C3=A9bastien=20P=C3=A9dron?= Date: Tue, 25 Feb 2025 19:22:54 +0100 Subject: [PATCH 2/8] amqp10_client: Handle `close` message in the `open_sent` state [Why] Without this, the connection process crashes. We see this happenning in CI frequently. --- .../src/amqp10_client_connection.erl | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/deps/amqp10_client/src/amqp10_client_connection.erl b/deps/amqp10_client/src/amqp10_client_connection.erl index 764846a21ac4..fd9ac19d3636 100644 --- a/deps/amqp10_client/src/amqp10_client_connection.erl +++ b/deps/amqp10_client/src/amqp10_client_connection.erl @@ -287,6 +287,22 @@ open_sent({call, From}, begin_session, #state{pending_session_reqs = PendingSessionReqs} = State) -> State1 = State#state{pending_session_reqs = [From | PendingSessionReqs]}, {keep_state, State1}; +open_sent(_EvtType, {close, Reason}, State) -> + %% TODO: stop all sessions writing + %% We could still accept incoming frames (See: 2.4.6) + case send_close(State, Reason) of + ok -> + %% "After writing this frame the peer SHOULD continue to read from the connection + %% until it receives the partner's close frame (in order to guard against + %% erroneously or maliciously implemented partners, a peer SHOULD implement a + %% timeout to give its partner a reasonable time to receive and process the close + %% before giving up and simply closing the underlying transport mechanism)." [ยง2.4.3] + {next_state, close_sent, State, {state_timeout, ?TIMEOUT, received_no_close_frame}}; + {error, closed} -> + {stop, normal, State}; + Error -> + {stop, Error, State} + end; open_sent(info, {'DOWN', MRef, process, _, _}, #state{reader_m_ref = MRef}) -> {stop, {shutdown, reader_down}}. From 65576863fc93cbf503db870bb4468923bfbd831b Mon Sep 17 00:00:00 2001 From: David Ansari Date: Tue, 4 Mar 2025 10:44:21 +0100 Subject: [PATCH 3/8] amqp10_client: Fix crash in close_sent Fix crash in close_sent since the client might receive the open frame if it previously sent the close frame in state open_sent. We explicitly ignore the open frame. The alternative is to add another gen_statem state CLOSE_PIPE which might be an overkill however. This commit also fixes a wrong comment: No sessions have begun if the app requests the connection to be closed in state open_sent. --- deps/amqp10_client/src/amqp10_client_connection.erl | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/deps/amqp10_client/src/amqp10_client_connection.erl b/deps/amqp10_client/src/amqp10_client_connection.erl index fd9ac19d3636..89a3396d85c1 100644 --- a/deps/amqp10_client/src/amqp10_client_connection.erl +++ b/deps/amqp10_client/src/amqp10_client_connection.erl @@ -288,8 +288,6 @@ open_sent({call, From}, begin_session, State1 = State#state{pending_session_reqs = [From | PendingSessionReqs]}, {keep_state, State1}; open_sent(_EvtType, {close, Reason}, State) -> - %% TODO: stop all sessions writing - %% We could still accept incoming frames (See: 2.4.6) case send_close(State, Reason) of ok -> %% "After writing this frame the peer SHOULD continue to read from the connection @@ -361,7 +359,10 @@ close_sent(_EvtType, #'v1_0.close'{} = Close, #state{config = Config}) -> ok = notify_closed(Config, Close), {stop, normal}; close_sent(state_timeout, received_no_close_frame, _Data) -> - {stop, normal}. + {stop, normal}; +close_sent(_EvtType, #'v1_0.open'{}, _Data) -> + %% Transition from CLOSE_PIPE to CLOSE_SENT in figure 2.23. + keep_state_and_data. set_other_procs0(OtherProcs, State) -> #{sessions_sup := SessionsSup, From 603ad0d7eb5edcea11b71c65e1b7341833bedbb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-S=C3=A9bastien=20P=C3=A9dron?= Date: Mon, 24 Feb 2025 16:04:00 +0100 Subject: [PATCH 4/8] amqp_client_SUITE: Retry connection in two testcases The testcases are `leader_transfer_credit` and `dead_letter_into_stream`. --- deps/rabbit/test/amqp_client_SUITE.erl | 29 ++++++++++++++++---------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl index 8beb7a6d458f..e29f3e19a1a3 100644 --- a/deps/rabbit/test/amqp_client_SUITE.erl +++ b/deps/rabbit/test/amqp_client_SUITE.erl @@ -11,6 +11,7 @@ -include_lib("eunit/include/eunit.hrl"). -include_lib("amqp_client/include/amqp_client.hrl"). -include_lib("amqp10_common/include/amqp10_framing.hrl"). +-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl"). -compile([nowarn_export_all, export_all]). @@ -3860,11 +3861,14 @@ leader_transfer_stream_credit_batches(Config) -> leader_transfer_credit(QName, QType, Credit, Config) -> %% Create queue with leader on node 1. {_, _, LinkPair1} = Init = init(1, Config), - {ok, #{type := QType}} = rabbitmq_amqp_client:declare_queue( - LinkPair1, - QName, - #{arguments => #{<<"x-queue-type">> => {utf8, QType}, - <<"x-queue-leader-locator">> => {utf8, <<"client-local">>}}}), + ?awaitMatch( + {ok, #{type := QType}}, + rabbitmq_amqp_client:declare_queue( + LinkPair1, + QName, + #{arguments => #{<<"x-queue-type">> => {utf8, QType}, + <<"x-queue-leader-locator">> => {utf8, <<"client-local">>}}}), + 60000), ok = close(Init), OpnConf = connection_config(0, Config), @@ -5436,12 +5440,15 @@ dead_letter_into_stream(Config) -> <<"x-dead-letter-exchange">> => {utf8, <<>>}, <<"x-dead-letter-routing-key">> => {utf8, QName1} }}), - {ok, #{type := <<"stream">>}} = rabbitmq_amqp_client:declare_queue( - LinkPair1, - QName1, - #{arguments => #{<<"x-queue-type">> => {utf8, <<"stream">>}, - <<"x-initial-cluster-size">> => {ulong, 1} - }}), + ?awaitMatch( + {ok, #{type := <<"stream">>}}, + rabbitmq_amqp_client:declare_queue( + LinkPair1, + QName1, + #{arguments => #{<<"x-queue-type">> => {utf8, <<"stream">>}, + <<"x-initial-cluster-size">> => {ulong, 1} + }}), + 60000), {ok, Receiver} = amqp10_client:attach_receiver_link( Session1, <<"receiver">>, <<"/amq/queue/", QName1/binary>>, settled, configuration, From 60840551831679f92c4c6365130645318bc3dfa4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-S=C3=A9bastien=20P=C3=A9dron?= Date: Mon, 24 Feb 2025 17:25:09 +0100 Subject: [PATCH 5/8] amqp_client_SUITE: Ensure `idle_time_out_on_server` restores heartbeat value [Why] If the testcase fails, it was leaving the low heartbeat value in place, leading to many subsequent tests to fail. --- deps/rabbit/test/amqp_client_SUITE.erl | 74 +++++++++++++------------- 1 file changed, 38 insertions(+), 36 deletions(-) diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl index e29f3e19a1a3..e7416c719ec1 100644 --- a/deps/rabbit/test/amqp_client_SUITE.erl +++ b/deps/rabbit/test/amqp_client_SUITE.erl @@ -4610,43 +4610,45 @@ plugin(Config) -> idle_time_out_on_server(Config) -> App = rabbit, Par = heartbeat, - {ok, DefaultVal} = rpc(Config, application, get_env, [App, Par]), - %% Configure RabbitMQ to use an idle-time-out of 1 second. - ok = rpc(Config, application, set_env, [App, Par, 1]), - - OpnConf = connection_config(Config), - {ok, Connection} = amqp10_client:open_connection(OpnConf), - receive {amqp10_event, {connection, Connection, opened}} -> ok - after 30000 -> ct:fail({missing_event, ?LINE}) - end, - - %% Mock the server socket to not have received any bytes. - rabbit_ct_broker_helpers:setup_meck(Config), Mod = rabbit_net, - ok = rpc(Config, meck, new, [Mod, [no_link, passthrough]]), - ok = rpc(Config, meck, expect, [Mod, getstat, fun(_Sock, [recv_oct]) -> - {ok, [{recv_oct, 999}]}; - (Sock, Opts) -> - meck:passthrough([Sock, Opts]) - end]), - - %% The server "SHOULD try to gracefully close the connection using a close - %% frame with an error explaining why" [2.4.5]. - %% Since we chose a heartbeat value of 1 second, the server should easily - %% close the connection within 5 seconds. - receive - {amqp10_event, - {connection, Connection, - {closed, - {resource_limit_exceeded, - <<"no frame received from client within idle timeout threshold">>}}}} -> ok - after 30000 -> - ct:fail({missing_event, ?LINE}) - end, - - ?assert(rpc(Config, meck, validate, [Mod])), - ok = rpc(Config, meck, unload, [Mod]), - ok = rpc(Config, application, set_env, [App, Par, DefaultVal]). + {ok, DefaultVal} = rpc(Config, application, get_env, [App, Par]), + try + %% Configure RabbitMQ to use an idle-time-out of 1 second. + ok = rpc(Config, application, set_env, [App, Par, 1]), + + OpnConf = connection_config(Config), + {ok, Connection} = amqp10_client:open_connection(OpnConf), + receive {amqp10_event, {connection, Connection, opened}} -> ok + after 30000 -> ct:fail({missing_event, ?LINE}) + end, + + %% Mock the server socket to not have received any bytes. + rabbit_ct_broker_helpers:setup_meck(Config), + ok = rpc(Config, meck, new, [Mod, [no_link, passthrough]]), + ok = rpc(Config, meck, expect, [Mod, getstat, fun(_Sock, [recv_oct]) -> + {ok, [{recv_oct, 999}]}; + (Sock, Opts) -> + meck:passthrough([Sock, Opts]) + end]), + + %% The server "SHOULD try to gracefully close the connection using a close + %% frame with an error explaining why" [2.4.5]. + %% Since we chose a heartbeat value of 1 second, the server should easily + %% close the connection within 5 seconds. + receive + {amqp10_event, + {connection, Connection, + {closed, + {resource_limit_exceeded, + <<"no frame received from client within idle timeout threshold">>}}}} -> ok + after 30000 -> + ct:fail({missing_event, ?LINE}) + end + after + ?assert(rpc(Config, meck, validate, [Mod])), + ok = rpc(Config, meck, unload, [Mod]), + ok = rpc(Config, application, set_env, [App, Par, DefaultVal]) + end. %% Test that the idle timeout threshold is exceeded on the client %% when no frames are sent from server to client. From ce5ba6da04119bc648f328e6ce293ef5ad2059b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-S=C3=A9bastien=20P=C3=A9dron?= Date: Wed, 26 Feb 2025 23:45:20 +0100 Subject: [PATCH 6/8] amqp_client_SUITE: Use a dedicated AMQP-0-9-1 connection per testcase ... instead of a global one. Otherwise, one connection failure, even if expected by a testcase, will affect all subsequent testcases negatively. --- deps/rabbit/test/amqp_client_SUITE.erl | 129 +++++++++++++------------ 1 file changed, 67 insertions(+), 62 deletions(-) diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl index e7416c719ec1..8f666adf2b0b 100644 --- a/deps/rabbit/test/amqp_client_SUITE.erl +++ b/deps/rabbit/test/amqp_client_SUITE.erl @@ -587,7 +587,7 @@ modified_quorum_queue(Config) -> ok = amqp10_client:settle_msg(Receiver1, M2e, modified), %% Test that we can consume via AMQP 0.9.1 - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), {#'basic.get_ok'{}, #amqp_msg{payload = <<"m2">>, props = #'P_basic'{headers = Headers}} @@ -598,7 +598,7 @@ modified_quorum_queue(Config) -> lists:keysearch(<<"x-other">>, 1, Headers)), ?assertEqual({value, {<<"x-delivery-count">>, long, 5}}, lists:keysearch(<<"x-delivery-count">>, 1, Headers)), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = amqp10_client:detach_link(Receiver1), {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QName), @@ -1344,7 +1344,7 @@ amqp_amqpl(QType, Config) -> ok = amqp10_client:detach_link(Sender), flush(detached), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'basic.qos_ok'{} = amqp_channel:call(Ch, #'basic.qos'{global = false, prefetch_count = 100}), CTag = <<"my-tag">>, @@ -1427,7 +1427,7 @@ amqp_amqpl(QType, Config) -> after 30000 -> ct:fail({missing_deliver, ?LINE}) end, - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QName), ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair), ok = close_connection_sync(Connection). @@ -1436,7 +1436,7 @@ message_headers_conversion(Config) -> QName = atom_to_binary(?FUNCTION_NAME), Address = rabbitmq_amqp_address:queue(QName), %% declare a quorum queue - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), amqp_channel:call(Ch, #'queue.declare'{ queue = QName, durable = true, @@ -1448,7 +1448,7 @@ message_headers_conversion(Config) -> amqp10_to_amqp091_header_conversion(Session, Ch, QName, Address), amqp091_to_amqp10_header_conversion(Session, Ch, QName, Address), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = delete_queue(Session, QName), ok = close_connection_sync(Connection). @@ -1554,11 +1554,11 @@ multiple_sessions(Config) -> ok = amqp10_client:flow_link_credit(Receiver2, NMsgsPerReceiver, never), flush("receiver attached"), - Ch = rabbit_ct_client_helpers:open_channel(Config, 0), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), [#'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue = QName, exchange = <<"amq.fanout">>}) || QName <- Qs], - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), %% Send on each session. TargetAddr = rabbitmq_amqp_address:exchange(<<"amq.fanout">>), @@ -1614,13 +1614,13 @@ server_closes_link_stream(Config) -> server_closes_link(QType, Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, durable = true, arguments = [{<<"x-queue-type">>, longstr, QType}]}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), @@ -1695,7 +1695,7 @@ server_closes_link_exchange(Settled, Config) -> XName = atom_to_binary(?FUNCTION_NAME), QName = <<"my queue">>, RoutingKey = <<"my routing key">>, - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = XName}), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue = QName, @@ -1737,7 +1737,7 @@ server_closes_link_exchange(Settled, Config) -> ?assertMatch(#{publishers := 0}, get_global_counters(Config)), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = end_session_sync(Session), ok = close_connection_sync(Connection). @@ -1749,13 +1749,13 @@ link_target_quorum_queue_deleted(Config) -> link_target_queue_deleted(QType, Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, durable = true, arguments = [{<<"x-queue-type">>, longstr, QType}]}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), @@ -1810,7 +1810,7 @@ target_queues_deleted_accepted(Config) -> Q2 = <<"q2">>, Q3 = <<"q3">>, QNames = [Q1, Q2, Q3], - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), [begin #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue = QName, @@ -1859,7 +1859,7 @@ target_queues_deleted_accepted(Config) -> ?assertEqual(#'queue.delete_ok'{message_count = 2}, amqp_channel:call(Ch, #'queue.delete'{queue = Q1})), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ?assert(rpc(Config, meck, validate, [Mod])), ok = rpc(Config, meck, unload, [Mod]), ok = end_session_sync(Session), @@ -1944,7 +1944,7 @@ sync_get_unsettled_stream(Config) -> sync_get_unsettled(QType, Config) -> SenderSettleMode = unsettled, QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, @@ -2033,7 +2033,7 @@ sync_get_unsettled(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). sync_get_unsettled_2_classic_queue(Config) -> sync_get_unsettled_2(<<"classic">>, Config). @@ -2048,7 +2048,7 @@ sync_get_unsettled_2_stream(Config) -> sync_get_unsettled_2(QType, Config) -> SenderSettleMode = unsettled, QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, @@ -2123,7 +2123,7 @@ sync_get_unsettled_2(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). sync_get_settled_classic_queue(Config) -> sync_get_settled(<<"classic">>, Config). @@ -2138,7 +2138,7 @@ sync_get_settled_stream(Config) -> sync_get_settled(QType, Config) -> SenderSettleMode = settled, QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, @@ -2203,7 +2203,7 @@ sync_get_settled(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). timed_get_classic_queue(Config) -> timed_get(<<"classic">>, Config). @@ -2217,7 +2217,7 @@ timed_get_stream(Config) -> %% Synchronous get with a timeout, figure 2.44. timed_get(QType, Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, @@ -2275,7 +2275,7 @@ timed_get(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). stop_classic_queue(Config) -> stop(<<"classic">>, Config). @@ -2288,7 +2288,7 @@ stop_stream(Config) -> %% Test stopping a link, figure 2.46. stop(QType, Config) -> - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), QName = atom_to_binary(?FUNCTION_NAME), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ @@ -2354,7 +2354,7 @@ stop(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). consumer_priority_classic_queue(Config) -> consumer_priority(<<"classic">>, Config). @@ -2832,7 +2832,7 @@ detach_requeues_one_session_quorum_queue(Config) -> detach_requeue_one_session(QType, Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, @@ -2910,7 +2910,7 @@ detach_requeue_one_session(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{message_count = 0} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). detach_requeues_drop_head_classic_queue(Config) -> QName1 = <<"q1">>, @@ -3080,7 +3080,7 @@ detach_requeues_two_connections(QType, Config) -> resource_alarm_before_session_begin(Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), @@ -3131,11 +3131,11 @@ resource_alarm_before_session_begin(Config) -> ok = end_session_sync(Session1), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). resource_alarm_after_session_begin(Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), Address = rabbitmq_amqp_address:queue(QName), OpnConf = connection_config(Config), @@ -3198,13 +3198,13 @@ resource_alarm_after_session_begin(Config) -> ok = close_connection_sync(Connection1), ok = close_connection_sync(Connection2), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). %% Test case for %% https://github.com/rabbitmq/rabbitmq-server/issues/12816 resource_alarm_send_many(Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), Address = rabbitmq_amqp_address:queue(QName), OpnConf = connection_config(Config), @@ -3234,7 +3234,7 @@ resource_alarm_send_many(Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). auth_attempt_metrics(Config) -> open_and_close_connection(Config), @@ -3267,7 +3267,7 @@ max_message_size_client_to_server(Config) -> ok = rpc(Config, persistent_term, put, [max_message_size, MaxMessageSize]), QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), Address = rabbitmq_amqp_address:queue(QName), OpnConf = connection_config(Config), @@ -3291,12 +3291,12 @@ max_message_size_client_to_server(Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = rpc(Config, persistent_term, put, [max_message_size, DefaultMaxMessageSize]). max_message_size_server_to_client(Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), Address = rabbitmq_amqp_address:queue(QName), OpnConf = connection_config(Config), @@ -3345,13 +3345,13 @@ max_message_size_server_to_client(Config) -> ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). last_queue_confirms(Config) -> ClassicQ = <<"my classic queue">>, QuorumQ = <<"my quorum queue">>, Qs = [ClassicQ, QuorumQ], - Ch = rabbit_ct_client_helpers:open_channel(Config, 0), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{queue = ClassicQ}), #'queue.declare_ok'{} = amqp_channel:call( @@ -3417,13 +3417,13 @@ last_queue_confirms(Config) -> amqp_channel:call(Ch, #'queue.delete'{queue = ClassicQ})), ?assertEqual(#'queue.delete_ok'{message_count = 2}, amqp_channel:call(Ch, #'queue.delete'{queue = QuorumQ})), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). target_queue_deleted(Config) -> ClassicQ = <<"my classic queue">>, QuorumQ = <<"my quorum queue">>, Qs = [ClassicQ, QuorumQ], - Ch = rabbit_ct_client_helpers:open_channel(Config, 0), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{queue = ClassicQ}), #'queue.declare_ok'{} = amqp_channel:call( @@ -3489,11 +3489,12 @@ target_queue_deleted(Config) -> ok = close_connection_sync(Connection), ?assertEqual(#'queue.delete_ok'{message_count = 2}, amqp_channel:call(Ch, #'queue.delete'{queue = QuorumQ})), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). target_classic_queue_down(Config) -> ClassicQueueNode = 2, - Ch = rabbit_ct_client_helpers:open_channel(Config, ClassicQueueNode), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel( + Config, ClassicQueueNode), QName = atom_to_binary(?FUNCTION_NAME), Address = rabbitmq_amqp_address:queue(QName), #'queue.declare_ok'{} = amqp_channel:call( @@ -3501,7 +3502,7 @@ target_classic_queue_down(Config) -> queue = QName, durable = true, arguments = [{<<"x-queue-type">>, longstr, <<"classic">>}]}), - ok = rabbit_ct_client_helpers:close_channels_and_connection(Config, ClassicQueueNode), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), @@ -3579,7 +3580,8 @@ async_notify_unsettled_stream(Config) -> %% Test asynchronous notification, figure 2.45. async_notify(SenderSettleMode, QType, Config) -> %% Place queue leader on the old node. - Ch = rabbit_ct_client_helpers:open_channel(Config, 1), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel( + Config, 1), QName = atom_to_binary(?FUNCTION_NAME), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ @@ -3636,7 +3638,7 @@ async_notify(SenderSettleMode, QType, Config) -> end, #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = end_session_sync(Session), ok = close_connection_sync(Connection). @@ -3644,7 +3646,7 @@ async_notify(SenderSettleMode, QType, Config) -> %% (slow queue) does not impact other link receivers (fast queues) on the **same** session. %% (This is unlike AMQP legacy where a single slow queue will block the entire connection.) link_flow_control(Config) -> - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), CQ = <<"cq">>, QQ = <<"qq">>, #'queue.declare_ok'{} = amqp_channel:call( @@ -3657,6 +3659,7 @@ link_flow_control(Config) -> queue = QQ, durable = true, arguments = [{<<"x-queue-type">>, longstr, <<"quorum">>}]}), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), {ok, Session} = amqp10_client:begin_session_sync(Connection), @@ -3744,7 +3747,8 @@ quorum_queue_on_new_node(Config) -> %% In mixed version tests, run the queue leader with old code %% and queue client with new code, or vice versa. queue_and_client_different_nodes(QueueLeaderNode, ClientNode, QueueType, Config) -> - Ch = rabbit_ct_client_helpers:open_channel(Config, QueueLeaderNode), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel( + Config, QueueLeaderNode), QName = atom_to_binary(?FUNCTION_NAME), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{queue = QName, @@ -3813,7 +3817,7 @@ queue_and_client_different_nodes(QueueLeaderNode, ClientNode, QueueType, Config) ExpectedReadyMsgs = 0, ?assertEqual(#'queue.delete_ok'{message_count = ExpectedReadyMsgs}, amqp_channel:call(Ch, #'queue.delete'{queue = QName})), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = close_connection_sync(Connection). maintenance(Config) -> @@ -4013,7 +4017,7 @@ global_counters(Config) -> messages_redelivered_total := QQRedelivered0, messages_acknowledged_total := QQAcknowledged0} = get_global_counters(Config, rabbit_quorum_queue), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), CQ = <<"my classic queue">>, QQ = <<"my quorum queue">>, CQAddress = rabbitmq_amqp_address:queue(CQ), @@ -4138,7 +4142,7 @@ global_counters(Config) -> %% m4 was returned ?assertEqual(UnroutableReturned1 + 1, UnroutableReturned2), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = amqp10_client:detach_link(Sender), ok = end_session_sync(Session), ok = close_connection_sync(Connection). @@ -4146,12 +4150,12 @@ global_counters(Config) -> stream_bloom_filter(Config) -> Stream = atom_to_binary(?FUNCTION_NAME), Address = rabbitmq_amqp_address:queue(Stream), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), amqp_channel:call(Ch, #'queue.declare'{ queue = Stream, durable = true, arguments = [{<<"x-queue-type">>, longstr, <<"stream">>}]}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), @@ -4278,7 +4282,7 @@ available_messages_stream(Config) -> available_messages(QType, Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = QName, @@ -4370,7 +4374,7 @@ available_messages(QType, Config) -> ok = end_session_sync(Session), ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). incoming_message_interceptors(Config) -> Key = ?FUNCTION_NAME, @@ -4437,7 +4441,7 @@ trace(Q, QType, Config) -> RoutingKey = <<"my routing key">>, Payload = <<"my payload">>, CorrelationId = <<"my correlation ๐Ÿ‘€"/utf8>>, - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{ queue = Q, @@ -4516,6 +4520,7 @@ trace(Q, QType, Config) -> timer:sleep(20), ?assertMatch(#'basic.get_empty'{}, amqp_channel:call(Ch, #'basic.get'{queue = TraceQ})), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), ok = amqp10_client:detach_link(Sender), ok = amqp10_client:detach_link(Receiver), @@ -4560,9 +4565,9 @@ user_id(Config) -> message_ttl(Config) -> QName = atom_to_binary(?FUNCTION_NAME), Address = rabbitmq_amqp_address:queue(QName), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), {ok, Session} = amqp10_client:begin_session_sync(Connection), @@ -4747,7 +4752,7 @@ credential_expires(Config) -> %% Attaching to an exclusive source queue should fail. attach_to_exclusive_queue(Config) -> QName = <<"my queue">>, - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call( Ch, #'queue.declare'{queue = QName, durable = true, @@ -4770,7 +4775,7 @@ attach_to_exclusive_queue(Config) -> ok = close_connection_sync(Connection), #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch). + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch). dynamic_target_short_link_name(Config) -> OpnConf0 = connection_config(Config), @@ -5883,9 +5888,9 @@ receive_many_auto_flow(QType, Config) -> %% incoming-window being closed. incoming_window_closed_transfer_flow_order(Config) -> QName = atom_to_binary(?FUNCTION_NAME), - Ch = rabbit_ct_client_helpers:open_channel(Config), + {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}), - ok = rabbit_ct_client_helpers:close_channel(Ch), + ok = rabbit_ct_client_helpers:close_connection_and_channel(Conn, Ch), Address = rabbitmq_amqp_address:queue(QName), OpnConf = connection_config(Config), {ok, Connection} = amqp10_client:open_connection(OpnConf), From 4d12efae219204c6a2eb32c919bc524fbe720f75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-S=C3=A9bastien=20P=C3=A9dron?= Date: Fri, 28 Feb 2025 11:53:41 +0100 Subject: [PATCH 7/8] amqp_client_SUITE: Close all connections in `end_per_testcase/2` [Why] Many tests do not clean up their connections if they encounter a failure. This affects subsequent testcases negatively. --- deps/rabbit/test/amqp_client_SUITE.erl | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl index 8f666adf2b0b..e09879bff953 100644 --- a/deps/rabbit/test/amqp_client_SUITE.erl +++ b/deps/rabbit/test/amqp_client_SUITE.erl @@ -359,7 +359,11 @@ end_per_testcase(Testcase, Config) -> %% Assert that every testcase cleaned up. rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []), eventually(?_assertEqual([], rpc(Config, rabbit_amqqueue, list, []))), - %% Wait for sessions to terminate before starting the next test case. + %% Terminate all connections and wait for sessions to terminate before + %% starting the next test case. + _ = rabbit_ct_broker_helpers:rpc( + Config, 0, + rabbit_networking, close_all_connections, [<<"test finished">>]), eventually(?_assertEqual([], rpc(Config, rabbit_amqp_session, list_local, []))), %% Assert that global counters count correctly. eventually(?_assertMatch(#{publishers := 0, From 0f9b693ec45e396f11a50ee8aa1d6ecb4f497a53 Mon Sep 17 00:00:00 2001 From: David Ansari Date: Fri, 7 Mar 2025 16:49:11 +0100 Subject: [PATCH 8/8] Apply PR feedback --- deps/rabbit/test/amqp_client_SUITE.erl | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl index e09879bff953..6e75e9a8f1fe 100644 --- a/deps/rabbit/test/amqp_client_SUITE.erl +++ b/deps/rabbit/test/amqp_client_SUITE.erl @@ -356,14 +356,9 @@ init_per_testcase(Testcase, Config) -> rabbit_ct_helpers:testcase_started(Config, Testcase). end_per_testcase(Testcase, Config) -> - %% Assert that every testcase cleaned up. - rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []), - eventually(?_assertEqual([], rpc(Config, rabbit_amqqueue, list, []))), - %% Terminate all connections and wait for sessions to terminate before - %% starting the next test case. - _ = rabbit_ct_broker_helpers:rpc( - Config, 0, - rabbit_networking, close_all_connections, [<<"test finished">>]), + %% Clean up any queues, connections, and sessions. + rpc(Config, ?MODULE, delete_queues, []), + ok = rpc(Config, rabbit_networking, close_all_connections, [<<"test finished">>]), eventually(?_assertEqual([], rpc(Config, rabbit_amqp_session, list_local, []))), %% Assert that global counters count correctly. eventually(?_assertMatch(#{publishers := 0,