Increase the backoff strategy time (#345)

Gsantomaggio · web-flow · commit fa8f66bb5445 · 2024-01-19T15:36:49.000+01:00
* Increase the backoff strategy time

* The back-off reconnect strategy is increased sensibly.
  The previous value was too aggressive.
  It caused a lot of tentative and a lot of server requests.
  The TCP port can be ready during the restart, but the stream cannot be ready due to the sync.

* Introduce a random delay on the strategy to avoid having the same
  reconnection time in case the client has more producers and consumers.

* Introduce also a random delay on the disconnected part on disconnect for the same above reason.

---------

Signed-off-by: Gabriele Santomaggio &lt;G.santomaggio@gmail.com&gt;
diff --git a/RabbitMQ.Stream.Client/ClientExceptions.cs b/RabbitMQ.Stream.Client/ClientExceptions.cs
@@ -49,6 +49,15 @@ internal static bool IsStreamNotAvailable(Exception exception)
             return exception is CreateException { ResponseCode: ResponseCode.StreamNotAvailable };
         }
 
+        internal static void CheckLeader(StreamInfo metaStreamInfo)
+        {
+            if (metaStreamInfo.Leader.Equals(default(Broker)))
+            {
+                throw new LeaderNotFoundException(
+                    $"No leader found for streams {string.Join(" ", metaStreamInfo.Stream)}");
+            }
+        }
+
         public static void MaybeThrowException(ResponseCode responseCode, string message)
         {
             if (responseCode is ResponseCode.Ok)
diff --git a/RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt b/RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt
@@ -180,9 +180,6 @@ RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.ConnectionClosedHandler.get
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.ConnectionClosedHandler.set -> void
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.set -> void
-RabbitMQ.Stream.Client.Reconnect.IReconnectStrategy
-RabbitMQ.Stream.Client.Reconnect.IReconnectStrategy.WhenConnected(string itemIdentifier) -> System.Threading.Tasks.ValueTask
-RabbitMQ.Stream.Client.Reconnect.IReconnectStrategy.WhenDisconnected(string itemIdentifier) -> System.Threading.Tasks.ValueTask<bool>
 RabbitMQ.Stream.Client.Reliable.Consumer.Info.get -> RabbitMQ.Stream.Client.ConsumerInfo
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Crc32.get -> RabbitMQ.Stream.Client.ICrc32
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Crc32.set -> void
@@ -199,6 +196,9 @@ RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer.IsOpen() -> bool
 RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer.Send(ulong publishing, RabbitMQ.Stream.Client.Message message) -> System.Threading.Tasks.ValueTask
 RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig
 RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig.DeduplicatingProducerConfig(RabbitMQ.Stream.Client.StreamSystem streamSystem, string stream, string reference) -> void
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenConnected(string itemIdentifier) -> System.Threading.Tasks.ValueTask
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenDisconnected(string itemIdentifier) -> System.Threading.Tasks.ValueTask<bool>
 RabbitMQ.Stream.Client.Reliable.Producer.Info.get -> RabbitMQ.Stream.Client.ProducerInfo
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.get -> RabbitMQ.Stream.Client.ProducerFilter
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.set -> void
@@ -208,8 +208,8 @@ RabbitMQ.Stream.Client.Reliable.ProducerFactory._producer -> RabbitMQ.Stream.Cli
 RabbitMQ.Stream.Client.Reliable.ReliableBase.CompareStatus(RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus toTest) -> bool
 RabbitMQ.Stream.Client.Reliable.ReliableBase.UpdateStatus(RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus status) -> void
 RabbitMQ.Stream.Client.Reliable.ReliableBase._status -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
-RabbitMQ.Stream.Client.Reliable.ReliableConfig.ReconnectStrategy.get -> RabbitMQ.Stream.Client.Reconnect.IReconnectStrategy
-RabbitMQ.Stream.Client.Reliable.ReliableConfig.ResourceAvailableReconnectStrategy.get -> RabbitMQ.Stream.Client.Reconnect.IReconnectStrategy
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.ReconnectStrategy.get -> RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.ResourceAvailableReconnectStrategy.get -> RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
 RabbitMQ.Stream.Client.Reliable.ReliableConfig.ResourceAvailableReconnectStrategy.set -> void
 RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
 RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus.Closed = 3 -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
@@ -264,5 +264,6 @@ static RabbitMQ.Stream.Client.RawProducer.Create(RabbitMQ.Stream.Client.ClientPa
 static RabbitMQ.Stream.Client.RawSuperStreamConsumer.Create(RabbitMQ.Stream.Client.RawSuperStreamConsumerConfig rawSuperStreamConsumerConfig, System.Collections.Generic.IDictionary<string, RabbitMQ.Stream.Client.StreamInfo> streamInfos, RabbitMQ.Stream.Client.ClientParameters clientParameters, Microsoft.Extensions.Logging.ILogger logger = null) -> RabbitMQ.Stream.Client.ISuperStreamConsumer
 static RabbitMQ.Stream.Client.RawSuperStreamProducer.Create(RabbitMQ.Stream.Client.RawSuperStreamProducerConfig rawSuperStreamProducerConfig, System.Collections.Generic.IDictionary<string, RabbitMQ.Stream.Client.StreamInfo> streamInfos, RabbitMQ.Stream.Client.ClientParameters clientParameters, Microsoft.Extensions.Logging.ILogger logger = null) -> RabbitMQ.Stream.Client.ISuperStreamProducer
 static RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer.Create(RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig producerConfig, Microsoft.Extensions.Logging.ILogger<RabbitMQ.Stream.Client.Reliable.Producer> logger = null) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer>
+static RabbitMQ.Stream.Client.Reliable.ReliableBase.RandomWait() -> System.Threading.Tasks.Task
 static RabbitMQ.Stream.Client.RoutingHelper<T>.LookupLeaderConnection(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.StreamInfo metaDataInfo, RabbitMQ.Stream.Client.ConnectionsPool pool, Microsoft.Extensions.Logging.ILogger logger = null) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IClient>
 static RabbitMQ.Stream.Client.RoutingHelper<T>.LookupRandomConnection(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.StreamInfo metaDataInfo, RabbitMQ.Stream.Client.ConnectionsPool pool, Microsoft.Extensions.Logging.ILogger logger = null) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IClient>
diff --git a/RabbitMQ.Stream.Client/RawSuperStreamConsumer.cs b/RabbitMQ.Stream.Client/RawSuperStreamConsumer.cs
@@ -95,7 +95,6 @@ private RawConsumerConfig FromStreamConfig(string stream)
                 }
 
                 consumer?.Dispose();
-                _streamInfos.Remove(stream);
 
                 if (_config.ConnectionClosedHandler != null)
                 {
@@ -117,7 +116,6 @@ private RawConsumerConfig FromStreamConfig(string stream)
             {
                 _consumers.TryRemove(update.Stream, out var consumer);
                 consumer?.Close();
-                _streamInfos.Remove(update.Stream);
                 if (_config.MetadataHandler != null)
                 {
                     await _config.MetadataHandler(update).ConfigureAwait(false);
@@ -159,12 +157,13 @@ private async Task StartConsumers()
 
     public async Task ReconnectPartition(StreamInfo streamInfo)
     {
+        ClientExceptions.CheckLeader(streamInfo);
         await _semaphoreSlim.WaitAsync().ConfigureAwait(false);
         try
         {
             _consumers.TryRemove(streamInfo.Stream, out var consumer);
             consumer?.Dispose();
-            _streamInfos.TryAdd(streamInfo.Stream, streamInfo); // add the new stream infos
+            _streamInfos[streamInfo.Stream] = streamInfo;
             await MaybeAddConsumer(streamInfo.Stream).ConfigureAwait(false);
         }
         finally
diff --git a/RabbitMQ.Stream.Client/RawSuperStreamProducer.cs b/RabbitMQ.Stream.Client/RawSuperStreamProducer.cs
@@ -163,6 +163,7 @@ private async Task<IProducer> MaybeAddAndGetProducer(string stream)
 
     public async Task ReconnectPartition(StreamInfo streamInfo)
     {
+        ClientExceptions.CheckLeader(streamInfo);
         await _semaphoreSlim.WaitAsync().ConfigureAwait(false);
         try
         {
diff --git a/RabbitMQ.Stream.Client/Reliable/Consumer.cs b/RabbitMQ.Stream.Client/Reliable/Consumer.cs
@@ -6,7 +6,6 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
-using RabbitMQ.Stream.Client.Reconnect;
 
 namespace RabbitMQ.Stream.Client.Reliable;
 
diff --git a/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs b/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs
@@ -122,6 +122,7 @@ private async Task<IConsumer> SuperConsumer(bool boot)
                     OffsetSpec = offsetSpecs,
                     ConnectionClosedHandler = async (closeReason, partitionStream) =>
                     {
+                        await RandomWait().ConfigureAwait(false);
                         if (closeReason == ConnectionClosedReason.Normal)
                         {
                             BaseLogger.LogInformation("{Identity} is closed normally", ToString());
@@ -134,6 +135,7 @@ await OnEntityClosed(_consumerConfig.StreamSystem, partitionStream, r)
                     },
                     MetadataHandler = async update =>
                     {
+                        await RandomWait().ConfigureAwait(false);
                         var r = ((RawSuperStreamConsumer)(_consumer)).ReconnectPartition;
                         await OnEntityClosed(_consumerConfig.StreamSystem, update.Stream, r)
                             .ConfigureAwait(false);
diff --git a/RabbitMQ.Stream.Client/Reliable/IReconnectStrategy.cs b/RabbitMQ.Stream.Client/Reliable/IReconnectStrategy.cs
@@ -7,7 +7,7 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 
-namespace RabbitMQ.Stream.Client.Reconnect;
+namespace RabbitMQ.Stream.Client.Reliable;
 
 /// <summary>
 /// IReconnectStrategy is the interface to reconnect the TCP client
@@ -48,36 +48,24 @@ public BackOffReconnectStrategy(ILogger logger = null)
     // else the backoff will be too long
     private void MaybeResetTentatives()
     {
-        if (Tentatives > 5)
+        if (Tentatives > 4)
         {
             Tentatives = 1;
         }
     }
 
     public async ValueTask<bool> WhenDisconnected(string connectionIdentifier)
     {
-
         Tentatives <<= 1;
+        var next = Random.Shared.Next(Tentatives * 1000, Tentatives * 3000);
         _logger.LogInformation(
             "{ConnectionIdentifier} disconnected, check if reconnection needed in {ReconnectionDelayMs} ms",
             connectionIdentifier,
-            Tentatives * 100
+            next
         );
-        await Task.Delay(TimeSpan.FromMilliseconds(Tentatives * 100)).ConfigureAwait(false);
+        await Task.Delay(TimeSpan.FromMilliseconds(next)).ConfigureAwait(false);
         MaybeResetTentatives();
         return true;
-        // this will be in another commit
-        // Tentatives <<= 1;
-        // var next = Random.Shared.Next(Tentatives * 1000, Tentatives * 2000);
-        // _logger.LogInformation(
-        //     "{ConnectionIdentifier} disconnected, check if reconnection needed in {ReconnectionDelayMs} ms",
-        //     connectionIdentifier,
-        //     next
-        // );
-        //
-        // await Task.Delay(TimeSpan.FromMilliseconds(next)).ConfigureAwait(false);
-        // MaybeResetTentatives();
-        // return true;
     }
 
     public ValueTask WhenConnected(string connectionIdentifier)
@@ -112,7 +100,7 @@ public async ValueTask<bool> WhenDisconnected(string resourceIdentifier)
     {
         Tentatives <<= 1;
         _logger.LogInformation(
-            "{ConnectionIdentifier} resource not available, retry in {ReconnectionDelayMs} seconds",
+            "{ConnectionIdentifier} resource not available, retry in {ReconnectionDelayS} seconds",
             resourceIdentifier,
             Tentatives
         );
diff --git a/RabbitMQ.Stream.Client/Reliable/Producer.cs b/RabbitMQ.Stream.Client/Reliable/Producer.cs
@@ -9,7 +9,6 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
-using RabbitMQ.Stream.Client.Reconnect;
 
 namespace RabbitMQ.Stream.Client.Reliable;
 
diff --git a/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs b/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs
@@ -47,6 +47,7 @@ private async Task<IProducer> SuperStreamProducer(bool boot)
                     Filter = _producerConfig.Filter,
                     ConnectionClosedHandler = async (closeReason, partitionStream) =>
                     {
+                        await RandomWait().ConfigureAwait(false);
                         if (closeReason == ConnectionClosedReason.Normal)
                         {
                             BaseLogger.LogDebug("{Identity} is closed normally", ToString());
@@ -59,6 +60,7 @@ await OnEntityClosed(_producerConfig.StreamSystem, partitionStream, r)
                     },
                     MetadataHandler = async update =>
                     {
+                        await RandomWait().ConfigureAwait(false);
                         var r = ((RawSuperStreamProducer)(_producer)).ReconnectPartition;
                         await OnEntityClosed(_producerConfig.StreamSystem, update.Stream, r)
                             .ConfigureAwait(false);
@@ -99,10 +101,12 @@ private async Task<IProducer> StandardProducer()
             Filter = _producerConfig.Filter,
             MetadataHandler = async _ =>
             {
+                await RandomWait().ConfigureAwait(false);
                 await OnEntityClosed(_producerConfig.StreamSystem, _producerConfig.Stream).ConfigureAwait(false);
             },
             ConnectionClosedHandler = async (closeReason) =>
             {
+                await RandomWait().ConfigureAwait(false);
                 if (closeReason == ConnectionClosedReason.Normal)
                 {
                     BaseLogger.LogDebug("{Identity} is closed normally", ToString());
diff --git a/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs b/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs
@@ -6,7 +6,6 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
-using RabbitMQ.Stream.Client.Reconnect;
 
 namespace RabbitMQ.Stream.Client.Reliable;
 
@@ -50,6 +49,11 @@ public abstract class ReliableBase
     private readonly object _lock = new();
     protected ReliableEntityStatus _status = ReliableEntityStatus.Initialization;
 
+    protected static async Task RandomWait()
+    {
+        await Task.Delay(Consts.RandomMid()).ConfigureAwait(false);
+    }
+
     protected void UpdateStatus(ReliableEntityStatus status)
     {
         lock (_lock)
@@ -112,7 +116,6 @@ private async Task MaybeInit(bool boot)
 
             reconnect = true;
             LogException(e);
-
         }
 
         if (reconnect)
@@ -130,6 +133,7 @@ private async Task Init(bool boot)
             BaseLogger.LogDebug("{Identity} is already closed. The init will be skipped", ToString());
             return;
         }
+
         // each time that the client is initialized, we need to reset the status
         // if we hare here it means that the entity is not open for some reason like:
         // first time initialization or reconnect due of a IsAKnownException
@@ -156,7 +160,8 @@ private async Task Init(bool boot)
     /// <summary>
     /// When the clients receives a meta data update, it doesn't know
     /// If the stream exists or not. It just knows that the stream topology has changed.
-    /// the method CheckIfStreamIsAvailable checks if the stream exists.
+    /// the method CheckIfStreamIsAvailable checks if the stream exists
+    /// and if the leader is available.
     /// </summary>
     /// <param name="stream">stream name</param>
     /// <param name="system">stream system</param>
@@ -172,7 +177,16 @@ private async Task<bool> CheckIfStreamIsAvailable(string stream, StreamSystem sy
             {
                 exists = await system.StreamExists(stream).ConfigureAwait(false);
                 var available = exists ? "available" : "not available";
-                await _resourceAvailableReconnectStrategy.WhenConnected($"{stream} is {available}")
+                if (exists)
+                {
+                    // It is not enough to check if the stream exists
+                    // we need to check if the stream has the leader
+                    var streamInfo = await system.StreamInfo(stream).ConfigureAwait(false);
+                    ClientExceptions.CheckLeader(streamInfo);
+                    available += " and has a valid leader";
+                }
+
+                await _resourceAvailableReconnectStrategy.WhenConnected($"{stream} for {ToString()} is {available}")
                     .ConfigureAwait(false);
                 break;
             }
@@ -229,7 +243,8 @@ private async Task MaybeReconnect()
         }
     }
 
-    private async Task MaybeReconnectPartition(StreamInfo streamInfo, string info, Func<StreamInfo, Task> reconnectPartitionFunc)
+    private async Task MaybeReconnectPartition(StreamInfo streamInfo, string info,
+        Func<StreamInfo, Task> reconnectPartitionFunc)
     {
         var reconnect = await _reconnectStrategy
             .WhenDisconnected($"Super Stream partition: {streamInfo.Stream} for {info}").ConfigureAwait(false);
@@ -285,7 +300,8 @@ private void LogException(Exception exception)
     /// <param name="system">Stream System</param>
     /// <param name="stream">Partition Stream</param>
     /// <param name="reconnectPartitionFunc">Function to reconnect the partition</param>
-    internal async Task OnEntityClosed(StreamSystem system, string stream, Func<StreamInfo, Task> reconnectPartitionFunc)
+    internal async Task OnEntityClosed(StreamSystem system, string stream,
+        Func<StreamInfo, Task> reconnectPartitionFunc)
     {
         var streamExists = false;
         await SemaphoreSlim.WaitAsync().ConfigureAwait(false);
diff --git a/RabbitMQ.Stream.Client/StreamSystem.cs b/RabbitMQ.Stream.Client/StreamSystem.cs
@@ -153,15 +153,6 @@ private async Task MayBeReconnectLocator()
             }
         }
 
-        private static void CheckLeader(StreamInfo metaStreamInfo)
-        {
-            if (metaStreamInfo.Leader.Equals(default(Broker)))
-            {
-                throw new LeaderNotFoundException(
-                    $"No leader found for streams {string.Join(" ", metaStreamInfo.Stream)}");
-            }
-        }
-
         public async Task<ISuperStreamProducer> CreateRawSuperStreamProducer(
             RawSuperStreamProducerConfig rawSuperStreamProducerConfig, ILogger logger = null)
         {
@@ -197,6 +188,11 @@ public async Task<ISuperStreamProducer> CreateRawSuperStreamProducer(
                 streamInfos[partitionsStream] = await StreamInfo(partitionsStream).ConfigureAwait(false);
             }
 
+            foreach (var (_, value) in streamInfos)
+            {
+                ClientExceptions.CheckLeader(value);
+            }
+
             var r = RawSuperStreamProducer.Create(rawSuperStreamProducerConfig,
                 streamInfos,
                 _clientParameters with { ClientProvidedName = rawSuperStreamProducerConfig.ClientProvidedName },
@@ -273,7 +269,7 @@ public async Task<IProducer> CreateRawProducer(RawProducerConfig rawProducerConf
                 throw new CreateProducerException($"producer could not be created code: {metaStreamInfo.ResponseCode}");
             }
 
-            CheckLeader(metaStreamInfo);
+            ClientExceptions.CheckLeader(metaStreamInfo);
 
             try
             {
@@ -432,7 +428,7 @@ public async Task<IConsumer> CreateRawConsumer(RawConsumerConfig rawConsumerConf
                     metaStreamInfo.ResponseCode);
             }
 
-            CheckLeader(metaStreamInfo);
+            ClientExceptions.CheckLeader(metaStreamInfo);
 
             try
             {
diff --git a/Tests/ReliableTests.cs b/Tests/ReliableTests.cs
diff --git a/Tests/SuperStreamConsumerTests.cs b/Tests/SuperStreamConsumerTests.cs
diff --git a/Tests/SuperStreamProducerTests.cs b/Tests/SuperStreamProducerTests.cs
diff --git a/Tests/Utils.cs b/Tests/Utils.cs

Original file line number	Diff line number	Diff line change
`@@ -95,7 +95,6 @@ private RawConsumerConfig FromStreamConfig(string stream)`
`95`	`95`	`}`
`96`	`96`
`97`	`97`	`consumer?.Dispose();`
`98`		`- _streamInfos.Remove(stream);`
`99`	`98`
`100`	`99`	`if (_config.ConnectionClosedHandler != null)`
`101`	`100`	`{`
`@@ -117,7 +116,6 @@ private RawConsumerConfig FromStreamConfig(string stream)`
`117`	`116`	`{`
`118`	`117`	`_consumers.TryRemove(update.Stream, out var consumer);`
`119`	`118`	`consumer?.Close();`
`120`		`- _streamInfos.Remove(update.Stream);`
`121`	`119`	`if (_config.MetadataHandler != null)`
`122`	`120`	`{`
`123`	`121`	`await _config.MetadataHandler(update).ConfigureAwait(false);`
`@@ -159,12 +157,13 @@ private async Task StartConsumers()`
`159`	`157`
`160`	`158`	`public async Task ReconnectPartition(StreamInfo streamInfo)`
`161`	`159`	`{`
	`160`	`+ ClientExceptions.CheckLeader(streamInfo);`
`162`	`161`	`await _semaphoreSlim.WaitAsync().ConfigureAwait(false);`
`163`	`162`	`try`
`164`	`163`	`{`
`165`	`164`	`_consumers.TryRemove(streamInfo.Stream, out var consumer);`
`166`	`165`	`consumer?.Dispose();`
`167`		`- _streamInfos.TryAdd(streamInfo.Stream, streamInfo); // add the new stream infos`
	`166`	`+ _streamInfos[streamInfo.Stream] = streamInfo;`
`168`	`167`	`await MaybeAddConsumer(streamInfo.Stream).ConfigureAwait(false);`
`169`	`168`	`}`
`170`	`169`	`finally`
Original file line number	Diff line number	Diff line change
`@@ -163,6 +163,7 @@ private async Task<IProducer> MaybeAddAndGetProducer(string stream)`
`163`	`163`
`164`	`164`	`public async Task ReconnectPartition(StreamInfo streamInfo)`
`165`	`165`	`{`
	`166`	`+ ClientExceptions.CheckLeader(streamInfo);`
`166`	`167`	`await _semaphoreSlim.WaitAsync().ConfigureAwait(false);`
`167`	`168`	`try`
`168`	`169`	`{`
Original file line number	Diff line number	Diff line change
`@@ -122,6 +122,7 @@ private async Task<IConsumer> SuperConsumer(bool boot)`
`122`	`122`	`OffsetSpec = offsetSpecs,`
`123`	`123`	`ConnectionClosedHandler = async (closeReason, partitionStream) =>`
`124`	`124`	`{`
	`125`	`+ await RandomWait().ConfigureAwait(false);`
`125`	`126`	`if (closeReason == ConnectionClosedReason.Normal)`
`126`	`127`	`{`
`127`	`128`	`BaseLogger.LogInformation("{Identity} is closed normally", ToString());`
`@@ -134,6 +135,7 @@ await OnEntityClosed(_consumerConfig.StreamSystem, partitionStream, r)`
`134`	`135`	`},`
`135`	`136`	`MetadataHandler = async update =>`
`136`	`137`	`{`
	`138`	`+ await RandomWait().ConfigureAwait(false);`
`137`	`139`	`var r = ((RawSuperStreamConsumer)(_consumer)).ReconnectPartition;`
`138`	`140`	`await OnEntityClosed(_consumerConfig.StreamSystem, update.Stream, r)`
`139`	`141`	`.ConfigureAwait(false);`
Original file line number	Diff line number	Diff line change
`@@ -47,6 +47,7 @@ private async Task<IProducer> SuperStreamProducer(bool boot)`
`47`	`47`	`Filter = _producerConfig.Filter,`
`48`	`48`	`ConnectionClosedHandler = async (closeReason, partitionStream) =>`
`49`	`49`	`{`
	`50`	`+ await RandomWait().ConfigureAwait(false);`
`50`	`51`	`if (closeReason == ConnectionClosedReason.Normal)`
`51`	`52`	`{`
`52`	`53`	`BaseLogger.LogDebug("{Identity} is closed normally", ToString());`
`@@ -59,6 +60,7 @@ await OnEntityClosed(_producerConfig.StreamSystem, partitionStream, r)`
`59`	`60`	`},`
`60`	`61`	`MetadataHandler = async update =>`
`61`	`62`	`{`
	`63`	`+ await RandomWait().ConfigureAwait(false);`
`62`	`64`	`var r = ((RawSuperStreamProducer)(_producer)).ReconnectPartition;`
`63`	`65`	`await OnEntityClosed(_producerConfig.StreamSystem, update.Stream, r)`
`64`	`66`	`.ConfigureAwait(false);`
`@@ -99,10 +101,12 @@ private async Task<IProducer> StandardProducer()`
`99`	`101`	`Filter = _producerConfig.Filter,`
`100`	`102`	`MetadataHandler = async _ =>`
`101`	`103`	`{`
	`104`	`+ await RandomWait().ConfigureAwait(false);`
`102`	`105`	`await OnEntityClosed(_producerConfig.StreamSystem, _producerConfig.Stream).ConfigureAwait(false);`
`103`	`106`	`},`
`104`	`107`	`ConnectionClosedHandler = async (closeReason) =>`
`105`	`108`	`{`
	`109`	`+ await RandomWait().ConfigureAwait(false);`
`106`	`110`	`if (closeReason == ConnectionClosedReason.Normal)`
`107`	`111`	`{`
`108`	`112`	`BaseLogger.LogDebug("{Identity} is closed normally", ToString());`
Original file line number	Diff line number	Diff line change
`@@ -153,15 +153,6 @@ private async Task MayBeReconnectLocator()`
`153`	`153`	`}`
`154`	`154`	`}`
`155`	`155`
`156`		`- private static void CheckLeader(StreamInfo metaStreamInfo)`
`157`		`- {`
`158`		`- if (metaStreamInfo.Leader.Equals(default(Broker)))`
`159`		`- {`
`160`		`- throw new LeaderNotFoundException(`
`161`		`- $"No leader found for streams {string.Join(" ", metaStreamInfo.Stream)}");`
`162`		`- }`
`163`		`- }`
`164`		`-`
`165`	`156`	`public async Task<ISuperStreamProducer> CreateRawSuperStreamProducer(`
`166`	`157`	`RawSuperStreamProducerConfig rawSuperStreamProducerConfig, ILogger logger = null)`
`167`	`158`	`{`
`@@ -197,6 +188,11 @@ public async Task<ISuperStreamProducer> CreateRawSuperStreamProducer(`
`197`	`188`	`streamInfos[partitionsStream] = await StreamInfo(partitionsStream).ConfigureAwait(false);`
`198`	`189`	`}`
`199`	`190`
	`191`	`+ foreach (var (_, value) in streamInfos)`
	`192`	`+ {`
	`193`	`+ ClientExceptions.CheckLeader(value);`
	`194`	`+ }`
	`195`	`+`
`200`	`196`	`var r = RawSuperStreamProducer.Create(rawSuperStreamProducerConfig,`
`201`	`197`	`streamInfos,`
`202`	`198`	`_clientParameters with { ClientProvidedName = rawSuperStreamProducerConfig.ClientProvidedName },`
`@@ -273,7 +269,7 @@ public async Task<IProducer> CreateRawProducer(RawProducerConfig rawProducerConf`
`273`	`269`	`throw new CreateProducerException($"producer could not be created code: {metaStreamInfo.ResponseCode}");`
`274`	`270`	`}`
`275`	`271`
`276`		`- CheckLeader(metaStreamInfo);`
	`272`	`+ ClientExceptions.CheckLeader(metaStreamInfo);`
`277`	`273`
`278`	`274`	`try`
`279`	`275`	`{`
`@@ -432,7 +428,7 @@ public async Task<IConsumer> CreateRawConsumer(RawConsumerConfig rawConsumerConf`
`432`	`428`	`metaStreamInfo.ResponseCode);`
`433`	`429`	`}`
`434`	`430`
`435`		`- CheckLeader(metaStreamInfo);`
	`431`	`+ ClientExceptions.CheckLeader(metaStreamInfo);`
`436`	`432`
`437`	`433`	`try`
`438`	`434`	`{`