fix linter

Author: KirillKurdyukov

src/Ydb.Sdk/src/Services/Topic/Producer.cs

@@ -37,8 +37,12 @@ internal Producer(ProducerConfig producerConfig, ISerializer<TValue> serializer)
 
     internal async Task Initialize()
     {
-        var stream = _config.Driver.BidirectionalStreamCall(TopicService.StreamWriteMethod,
-            GrpcRequestSettings.DefaultInstance);
+        _logger.LogInformation("Producer session initialization started. ProducerConfig: {ProducerConfig}", _config);
+
+        var stream = _config.Driver.BidirectionalStreamCall(
+            TopicService.StreamWriteMethod,
+            GrpcRequestSettings.DefaultInstance
+        );
 
         var initRequest = new StreamWriteMessage.Types.InitRequest { Path = _config.TopicPath };
         if (_config.ProducerId != null)
@@ -54,7 +58,8 @@ internal async Task Initialize()
         await stream.Write(new MessageFromClient { InitRequest = initRequest });
         if (!await stream.MoveNextAsync())
         {
-            throw new YdbProducerException("Write stream is closed by YDB server");
+            throw new YdbProducerException(
+                $"Stream unexpectedly closed by YDB server. Current InitRequest: {initRequest}");
         }
 
         var receivedInitMessage = stream.Current;
@@ -139,26 +144,41 @@ public ProducerSession(
 
     internal async Task RunProcessingWriteAck(ConcurrentQueue<MessageSending> inFlightMessages)
     {
-        await foreach (var messageFromServer in _stream)
+        try
         {
-            var status = Status.FromProto(messageFromServer.Status, messageFromServer.Issues);
-
-            if (status.IsNotSuccess)
+            Logger.LogInformation("ProducerSession[{SessionId}] is running processing writeAck", SessionId);
+            
+            await foreach (var messageFromServer in _stream)
             {
-                Logger.LogWarning("");
-                return;
-            }
+                var status = Status.FromProto(messageFromServer.Status, messageFromServer.Issues);
 
-            foreach (var ack in messageFromServer.WriteResponse.Acks)
-            {
-                if (!inFlightMessages.TryDequeue(out var messageFromClient))
+                if (status.IsNotSuccess)
                 {
-                    break;
+                    Logger.LogWarning(
+                        "ProducerSession[{SessionId}] received unsuccessful status while processing writeAck: {Status}",
+                        SessionId, status);
+                    return;
                 }
 
-                messageFromClient.TaskCompletionSource.SetResult(new SendResult(ack));
+                foreach (var ack in messageFromServer.WriteResponse.Acks)
+                {
+                    if (!inFlightMessages.TryDequeue(out var messageFromClient))
+                    {
+                        break;
+                    }
+
+                    messageFromClient.TaskCompletionSource.SetResult(new SendResult(ack));
+                }
             }
         }
+        catch (Exception e)
+        {
+            Logger.LogError(e, "ProducerSession[{SessionId}] have error on processing writeAck", SessionId);
+        }
+        finally
+        {
+            ReconnectSession();
+        }
     }
 
     internal async Task Write(ConcurrentQueue<MessageSending> toSendBuffer,

src/Ydb.Sdk/src/Services/Topic/ProducerConfig.cs

@@ -1,3 +1,5 @@
+using System.Text;
+
 namespace Ydb.Sdk.Services.Topic;
 
 public class ProducerConfig
@@ -13,4 +15,21 @@ public ProducerConfig(Driver driver, string topicPath)
     public string? ProducerId { get; set; }
     public string? MessageGroupId { get; set; }
     public Codec Codec { get; set; } = Codec.Raw; // TODO Supported only Raw
+
+    public override string ToString()
+    {
+        var toString = new StringBuilder().Append("[TopicPath: ").Append(TopicPath);
+
+        if (ProducerId != null)
+        {
+            toString.Append(", ProducerId: ").Append(ProducerId);
+        }
+
+        if (MessageGroupId != null)
+        {
+            toString.Append(", MessageGroupId: ").Append(MessageGroupId);
+        }
+
+        return toString.Append(", Codec: ").Append(Codec).Append(']').ToString();
+    }
 }

src/Ydb.Sdk/src/Services/Topic/TopicSession.cs

@@ -5,26 +5,29 @@ namespace Ydb.Sdk.Services.Topic;
 internal abstract class TopicSession : IDisposable
 {
     private readonly Func<Task> _initialize;
- 
+
     protected readonly ILogger Logger;
     protected readonly string SessionId;
-    
+
     private int _isActive = 1;
 
     protected TopicSession(ILogger logger, string sessionId, Func<Task> initialize)
     {
         Logger = logger;
-        _initialize = initialize;
         SessionId = sessionId;
+        _initialize = initialize;
     }
 
     protected async void ReconnectSession()
     {
         if (Interlocked.CompareExchange(ref _isActive, 0, 1) == 0)
         {
-            return;    
+            Logger.LogWarning("The reconnect has already been launched");
+
+            return;
         }
 
+
         while (true)
         {
             try
@@ -38,9 +41,8 @@ protected async void ReconnectSession()
             }
         }
     }
-    
+
     public void Dispose()
     {
-        
     }
 }