Fix MQTT test flake

Prior to this commit test block_connack_timeout
flaked when 2 new ports got created instead of only 1
in line
```
[NewPort] = Ports -- Ports0,
```

This commit filters for tcp_inet ports.
This will always return the port of the new MQTT connection.

(cherry picked from commit bedcae1)

Author: ansd

deps/rabbitmq_mqtt/test/reader_SUITE.erl

@@ -92,7 +92,7 @@ end_per_testcase(Testcase, Config) ->
 
 block_connack_timeout(Config) ->
     P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),
-    Ports0 = rpc(Config, erlang, ports, []),
+    Ports = rpc(Config, erlang, ports, []),
 
     ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
     %% Let connection block.
@@ -107,34 +107,35 @@ block_connack_timeout(Config) ->
     unlink(Client),
     ClientMRef = monitor(process, Client),
     {error, connack_timeout} = emqtt:connect(Client),
-    receive
-        {'DOWN', ClientMRef, process, Client, connack_timeout} ->
-            ok
-    after 200 ->
-              ct:fail("missing connack_timeout in client")
+    receive {'DOWN', ClientMRef, process, Client, connack_timeout} -> ok
+    after 200 -> ct:fail("missing connack_timeout in client")
     end,
 
-    Ports = rpc(Config, erlang, ports, []),
-    %% Server creates 1 new port to handle our MQTT connection.
-    [NewPort] = Ports -- Ports0,
-    {connected, MqttReader} = rpc(Config, erlang, port_info, [NewPort, connected]),
+    MqttReader = rpc(Config, ?MODULE, mqtt_connection_pid, [Ports]),
     MqttReaderMRef = monitor(process, MqttReader),
 
     %% Unblock connection. CONNECT packet will be processed on the server.
     rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0.4]),
 
-    receive
-        {'DOWN', MqttReaderMRef, process, MqttReader, {shutdown, {socket_ends, einval}}} ->
-            %% We expect that MQTT reader process exits (without crashing)
-            %% because our client already disconnected.
-            ok
-    after 2000 ->
-              ct:fail("missing peername_not_known from server")
+    receive {'DOWN', MqttReaderMRef, process, MqttReader, {shutdown, {socket_ends, einval}}} ->
+                %% We expect that MQTT reader process exits (without crashing)
+                %% because our client already disconnected.
+                ok
+    after 2000 -> ct:fail("missing peername_not_known from server")
     end,
     %% Ensure that our client is not registered.
     ?assertEqual([], all_connection_pids(Config)),
     ok.
 
+mqtt_connection_pid(ExistingPorts) ->
+    NewPorts = erlang:ports() -- ExistingPorts,
+    %% Server creates 1 new TCP port to handle our MQTT connection.
+    [MqttConnectionPort] = lists:filter(fun(P) ->
+                                                erlang:port_info(P, name) =:= {name, "tcp_inet"}
+                                        end, NewPorts),
+    {connected, MqttConnectionPid} = erlang:port_info(MqttConnectionPort, connected),
+    MqttConnectionPid.
+
 handle_invalid_packets(Config) ->
     N = rpc(Config, ets, info, [connection_metrics, size]),
     P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),