Topic SDK: change type of ServerBytesSize to atomic (#24252)

Author: qyryq

.github/last_commit.txt

@@ -1 +1 @@
-2663e554d10f2d617bf96acaa684868685f3a8d7
+66e1273f270f166cf9a11288b0b2310f55e6de82

src/client/topic/impl/read_session_impl.h

@@ -365,7 +365,7 @@ class TDataDecompressionInfo : public std::enable_shared_from_this<TDataDecompre
     std::vector<TMessageMetaPtrVector> MessagesMeta;
     TCallbackContextPtr<UseMigrationProtocol> CbContext;
     bool DoDecompress;
-    i64 ServerBytesSize = 0;
+    std::atomic<i64> ServerBytesSize = 0;
     std::atomic<i64> SourceDataNotProcessed = 0;
     std::pair<size_t, size_t> CurrentDecompressingMessage = {0, 0}; // (Batch, Message)
     std::deque<TReadyMessageThreshold> ReadyThresholds;

src/client/topic/impl/read_session_impl.ipp

@@ -24,6 +24,7 @@
 #include <util/generic/yexception.h>
 #include <util/stream/mem.h>
 
+#include <atomic>
 #include <utility>
 #include <variant>
 
@@ -2984,7 +2985,7 @@ void TDataDecompressionInfo<UseMigrationProtocol>::OnDataDecompressed(i64 source
     if (auto session = CbContext->LockShared()) {
         // TODO (ildar-khisam@): distribute total ServerBytesSize in proportion of source size
         // Use CompressedDataSize, sourceSize, ServerBytesSize
-        session->OnDataDecompressed(sourceSize, estimatedDecompressedSize, decompressedSize, messagesCount, std::exchange(ServerBytesSize, 0));
+        session->OnDataDecompressed(sourceSize, estimatedDecompressedSize, decompressedSize, messagesCount, ServerBytesSize.exchange(0));
     }
 }