Merge pull request #13409 from rabbitmq/fix-test-flakes-in-amqp_client_SUITE

amqp_client_SUITE: Fix frequent test failures

Author: dumbbell

.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

deps/amqp10_client/src/amqp10_client_connection.erl

@@ -287,6 +287,20 @@ 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) ->
+    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}}.
@@ -345,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,

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)),)