Fix concurrent problem with tracking head

Author: Jroland

src/kafka-net/Common/ConcurrentCircularBuffer.cs

@@ -8,7 +8,7 @@ namespace KafkaNet.Common
     public class ConcurrentCircularBuffer<T> : IEnumerable<T>
     {
         private readonly int _maxSize;
-        private long _count = 0;
+        private long _count;
         private int _head = -1;
         readonly T[] _values;
 
@@ -30,13 +30,18 @@ public long Count
 
         public ConcurrentCircularBuffer<T> Enqueue(T obj)
         {
-            if (Interlocked.Increment(ref _head) > (_maxSize - 1))
-                Interlocked.Exchange(ref _head, 0);
+            var head = Interlocked.Increment(ref _head);
 
-            _values[_head] = obj;
+            if (head > _maxSize - 1)
+            {
+                Interlocked.Exchange(ref _head, head - _maxSize);
+                head = head - _maxSize;
+            }
+
+            _values[head] = obj;
 
-            Interlocked.Exchange(ref _count,
-                Math.Min(Interlocked.Increment(ref _count), _maxSize));
+            if (_count != _maxSize) //once we hit max size we dont need to track count.
+                Interlocked.Exchange(ref _count, Math.Min(Interlocked.Increment(ref _count), _maxSize));
 
             return this;
         }

src/kafka-tests/Unit/CircularBufferTests.cs

@@ -1,4 +1,5 @@
 using System.Linq;
+using System.Threading.Tasks;
 using KafkaNet.Common;
 using NUnit.Framework;
 
@@ -66,5 +67,14 @@ public void EnqueueShouldAddToFirstSlot()
             buffer.Enqueue(1);
             Assert.That(buffer.First(), Is.EqualTo(1));
         }
+
+        [Test]
+        public void ConcurrentEnqueueShouldNotLeakIndexPosition()
+        {
+            var buffer = new ConcurrentCircularBuffer<int>(10);
+            Parallel.For(0, 100000, i => buffer.Enqueue(i));
+            Assert.That(buffer.Count, Is.EqualTo(10));
+            
+        }
     }
 }