instead of reseting round robin after a sniff let it count onwards

Author: Mpdreamz

src/Elasticsearch.Net.Tests.Unit/Connection/ConcurrencyTests.cs

@@ -39,7 +39,8 @@ public ConcurrencyTests()
 			_connectionPool = new SniffingConnectionPool(_uris);
 			_config = new ConnectionConfiguration(_connectionPool)
 				.SnifsOnConnectionFault()
-				.SniffOnStartup();
+				.SniffOnStartup()
+				.SetMaxRetries(5);
 		}
 
 		private void ProvideTransport(AutoFake fake)
@@ -58,12 +59,9 @@ public void CallInfo40000TimesOnMultipleThreads()
 				//prove a real HttpTransport with its unspecified dependencies
 				//as fakes
 
-				//set up fake for a call on IConnection.GetSync so that it always throws 
-				//an exception
 				var connection = fake.Provide<IConnection>(new ConcurrencyTestConnection(this._config));
 				this.ProvideTransport(fake);
-				//create a real ElasticsearchClient with it unspecified dependencies
-				//as fakes
+				//create a real ElasticsearchClient with it unspecified dependencies as fakes
 				var client = fake.Resolve<ElasticsearchClient>();
 				int seen = 0;
 				Assert.DoesNotThrow(()=>
@@ -93,7 +91,14 @@ public void CallInfo40000TimesOnMultipleThreads()
 				seen.Should().Be(40000);
 			}
 		}
-
+		/// <summary>
+		/// This simulates a super flakey elasticsearch cluster.
+		/// - if random 1-9 is a muliple of 3 throw a 503
+		/// - never throws on node 9202 though so that all calls can be expected to always succeed. 
+		/// - Sniff can either get back the full cluster or a sufficient subset of it. 
+		/// - Our cluster have 5 nodes the recommendation is to have N/2+1 masters so we should atleast see 3 nodes
+		/// - anything less would cause a node to be unavailable which is covered in other tests 
+		/// </summary>
 		public class ConcurrencyTestConnection : InMemoryConnection
 		{
 			private static Uri[] _uris = new[]
@@ -102,6 +107,14 @@ public class ConcurrencyTestConnection : InMemoryConnection
 				new Uri("http://localhost:9201"),
 				new Uri("http://localhost:9202"),
 				new Uri("http://localhost:9203"),
+				new Uri("http://localhost:9206"),
+			};
+			
+			private static Uri[] _uris2 = new[]
+			{
+				new Uri("http://localhost:9202"),
+				new Uri("http://localhost:9201"),
+				new Uri("http://localhost:9206"),
 			};
 			private readonly Random _rnd = new Random();
 			public ConcurrencyTestConnection(IConnectionConfigurationValues settings) : base(settings)
@@ -110,12 +123,12 @@ public ConcurrencyTestConnection(IConnectionConfigurationValues settings) : base
 
 			public override IList<Uri> Sniff(Uri uri, int connectTimeout)
 			{
-				return _uris;
+				return  _rnd.Next(1, 11) % 3 == 0 ? _uris : _uris2;
 			}
 
 			public override ElasticsearchResponse GetSync(Uri uri)
 			{
-				var statusCode = _rnd.Next(1, 11) % 7 == 0 ? 503 : 200;
+				var statusCode = _rnd.Next(1, 9) % 3 == 0 ? 503 : 200;
 				if (uri.Port == 9202)
 					statusCode = 200;
 

src/Elasticsearch.Net.Tests.Unit/Connection/SniffingConnectionPoolTests.cs

@@ -205,7 +205,11 @@ public void HostsReturnedBySniffAreVisited()
 					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 2 retry
 					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 3
 					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 4
-					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null) //info 5
+					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 5
+					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 6
+					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 7
+					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null), //info 8
+					ElasticsearchResponse.Create(config, 200, "GET", "/", null, null) //info 9
 				);
 				getCall.Invokes((Uri u) => seenNodes.Add(u));
 
@@ -215,16 +219,21 @@ public void HostsReturnedBySniffAreVisited()
 				client1.Info(); //info call 3
 				client1.Info(); //info call 4
 				client1.Info(); //info call 5
+				client1.Info(); //info call 6
+				client1.Info(); //info call 7
+				client1.Info(); //info call 8
+				client1.Info(); //info call 9
 
 				sniffCall.MustHaveHappened(Repeated.Exactly.Once);
-				seenNodes.Should().NotBeEmpty().And.HaveCount(6);
+				seenNodes.Should().NotBeEmpty().And.HaveCount(10);
 				seenNodes[0].Port.Should().Be(9200);
 				seenNodes[1].Port.Should().Be(9201);
 				//after sniff
-				seenNodes[2].Port.Should().Be(9204);
-				seenNodes[3].Port.Should().Be(9203);
-				seenNodes[4].Port.Should().Be(9202);
-				seenNodes[5].Port.Should().Be(9201);
+				seenNodes[2].Port.Should().Be(9202);
+				seenNodes[3].Port.Should().Be(9204, string.Join(",", seenNodes.Select(n=>n.Port)));
+				seenNodes[4].Port.Should().Be(9203);
+				seenNodes[5].Port.Should().Be(9202);
+				seenNodes[6].Port.Should().Be(9201);
 
 				//var nowCall = A.CallTo(() => fake.Resolve<IDateTimeProvider>().Sniff(A<Uri>._, A<int>._));
 			}

src/Elasticsearch.Net/ConnectionPool/SniffingConnectionPool.cs

@@ -40,7 +40,6 @@ public override void Sniff(IConnection connection, bool fromStartupHint = false)
 				this._uriLookup = nodes.ToDictionary(k => k, v => new EndpointState());
 				if (fromStartupHint)
 					this._seenStartup = true;
-				this._current = -1;
 
 			}
 			finally

src/Elasticsearch.Net/ConnectionPool/StaticConnectionPool.cs

@@ -33,20 +33,15 @@ public StaticConnectionPool(
 
 		public virtual Uri GetNext(int? initialSeed, out int seed)
 		{
-			//if _current has been reset make sure we ignore any seed
-			//that was given out before the reset
-			if (_current == -1 && initialSeed.HasValue)
-				initialSeed = 0;
-			//always increment the initialSeed so we advance further
-			else if (initialSeed.HasValue)
+			var count = _nodeUris.Count;
+			if (initialSeed.HasValue)
 				initialSeed += 1;
 
 			//always increment our round robin counter
 			int increment = Interlocked.Increment(ref _current);
 			var initialOffset = initialSeed ?? increment;
-			seed = initialOffset;
-			var count = _nodeUris.Count;
 			int i = initialOffset % count, attempts = 0;
+			seed = i;
 			Uri uri = null;
 			do
 			{
@@ -94,7 +89,6 @@ public virtual void MarkAlive(Uri uri)
 
 		public virtual void Sniff(IConnection connection, bool fromStartupHint = false)
 		{
-			this._current = -1;
 			//NOOP on static connection class
 		}
 	}