Handle infinite looping when encountering a misconfigured load balancer (#121)

* Handle misconfigured load balancer

* Add logic to handle misconfigured load balancer

* Update LookupConnection to use maxAttempts as an argument

Signed-off-by: Gabriele Santomaggio <[email protected]>

Co-authored-by: Gabriele Santomaggio <[email protected]>

Author: ricsiLT

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -19,7 +19,6 @@ public interface IRouting
 
     public class Routing : IRouting
     {
-
         public bool ValidateDns { get; set; } = true;
 
         public IClient CreateClient(ClientParameters clientParameters)
@@ -39,8 +38,11 @@ public IClient CreateClient(ClientParameters clientParameters)
     /// </summary>
     public static class RoutingHelper<T> where T : IRouting, new()
     {
-        private static async Task<IClient> LookupConnection(ClientParameters clientParameters,
-            Broker broker)
+        private static async Task<IClient> LookupConnection(
+            ClientParameters clientParameters,
+            Broker broker,
+            int maxAttempts
+        )
         {
             var routing = new T();
 
@@ -80,29 +82,49 @@ string GetPropertyValue(IDictionary<string, string> connectionProperties, string
             var advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
             var advertisedPort = GetPropertyValue(client.ConnectionProperties, "advertised_port");
 
-            while (broker.Host != advertisedHost ||
-                   broker.Port != uint.Parse(advertisedPort))
+            var attemptNo = 0;
+            while (broker.Host != advertisedHost || broker.Port != uint.Parse(advertisedPort))
             {
-                await client.Close("advertised_host or advertised_port doesn't mach");
+                attemptNo++;
+                await client.Close("advertised_host or advertised_port doesn't match");
 
                 client = routing.CreateClient(clientParameters with { Endpoint = endPoint });
 
                 advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
                 advertisedPort = GetPropertyValue(client.ConnectionProperties, "advertised_port");
-                // TODO: Maybe an external parameter
+                if (attemptNo > maxAttempts)
+                {
+                    throw new RoutingClientException(
+                        $"Could not find broker ({broker.Host}:{broker.Port}) after {maxAttempts} attempts");
+                }
+
                 Thread.Sleep(TimeSpan.FromMilliseconds(200));
             }
 
             return client;
         }
 
+        private static int MaxAttempts(StreamInfo metaDataInfo)
+        {
+            // Here we have a reasonable number of retry.
+            // based on the stream configuration
+            // It will retry to the same node more than one time
+            // to be sure that there is not some temp fail
+            return (int)Math.Pow(2
+                                 +
+                                 1 // The leader node
+                                 +
+                                 metaDataInfo.Replicas.Count, // Replicas
+                2); // Pow just to be sure that the LoadBalancer will ping all the nodes
+        }
+
         /// <summary>
         /// Gets the leader connection. The producer must connect to the leader. 
         /// </summary>
         public static async Task<IClient> LookupLeaderConnection(ClientParameters clientParameters,
             StreamInfo metaDataInfo)
         {
-            return await LookupConnection(clientParameters, metaDataInfo.Leader);
+            return await LookupConnection(clientParameters, metaDataInfo.Leader, MaxAttempts(metaDataInfo));
         }
 
         /// <summary>
@@ -121,7 +143,8 @@ public static async Task<IClient> LookupRandomConnection(ClientParameters client
             var rnd = new Random();
             var brokerId = rnd.Next(0, brokers.Count);
             var broker = brokers[brokerId];
-            return await LookupConnection(clientParameters, broker);
+
+            return await LookupConnection(clientParameters, broker, MaxAttempts(metaDataInfo));
         }
     }
 

Tests/UnitTests.cs

@@ -59,6 +59,26 @@ public IClient CreateClient(ClientParameters clientParameters)
         public bool ValidateDns { get; set; } = false;
     }
 
+    public class MisconfiguredLoadBalancerRouting : IRouting
+    {
+
+        public IClient CreateClient(ClientParameters clientParameters)
+        {
+
+            var fake = new FakeClient(clientParameters)
+            {
+                ConnectionProperties = new Dictionary<string, string>()
+                {
+                    ["advertised_host"] = "node4",
+                    ["advertised_port"] = "5552"
+                }
+            };
+            return fake;
+        }
+
+        public bool ValidateDns { get; set; } = false;
+    }
+
     //advertised_host is is missed
     public class MissingFieldsRouting : IRouting
     {
@@ -132,6 +152,21 @@ public void AddressResolverLoadBalancerSimulate()
             }
         }
 
+        [Fact]
+        public void RoutingHelperShouldThrowIfLoadBalancerIsMisconfigured()
+        {
+            var addressResolver = new AddressResolver(new IPEndPoint(IPAddress.Parse("192.168.10.99"), 5552));
+            var clientParameters = new ClientParameters()
+            {
+                AddressResolver = addressResolver,
+            };
+            var metaDataInfo = new StreamInfo("stream", ResponseCode.Ok, new Broker("node2", 5552),
+                new List<Broker>() { new Broker("replica", 5552) });
+
+            Assert.ThrowsAsync<RoutingClientException>(
+                () => RoutingHelper<MisconfiguredLoadBalancerRouting>.LookupLeaderConnection(clientParameters, metaDataInfo));
+        }
+
         [Fact]
         public void RandomReplicaLeader()
         {