[-] the message does not store a pointer

Author: Alek5andr-Kotov

ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.cpp

@@ -41,13 +41,22 @@ TTxIdOpt GetTransactionId(const Ydb::Topic::StreamWriteMessage_WriteRequest& req
     return TTxId(tx.session(), tx.id());
 }
 
-TTxIdOpt GetTransactionId(const NTable::TTransaction* tx)
+TTxIdOpt GetTransactionId(const std::optional<TTransactionId>& tx)
 {
     if (!tx) {
         return std::nullopt;
     }
 
-    return TTxId(tx->GetSession().GetId(), tx->GetId());
+    return TTxId(tx->SessionId, tx->TxId);
+}
+
+std::optional<TTransactionId> MakeTransactionId(const NTable::TTransaction* tx)
+{
+    if (!tx) {
+        return std::nullopt;
+    }
+
+    return TTransactionId{tx->GetSession().GetId(), tx->GetId()};
 }
 
 }
@@ -645,7 +654,7 @@ void TWriteSessionImpl::WriteInternal(TContinuationToken&&, TWriteMessage&& mess
         CurrentBatch.Add(
                 seqNo, createdAtValue, message.Data, message.Codec, message.OriginalSize,
                 message.MessageMeta_,
-                message.GetTxPtr()
+                MakeTransactionId(message.GetTxPtr())
         );
 
         FlushWriteIfRequiredImpl();
@@ -1418,10 +1427,10 @@ size_t TWriteSessionImpl::WriteBatchImpl() {
             if (!currMessage.MessageMeta.empty()) {
                 OriginalMessagesToSend.emplace(id, createTs, datum.size(),
                                                std::move(currMessage.MessageMeta),
-                                               currMessage.Tx);
+                                               std::move(currMessage.Tx));
             } else {
                 OriginalMessagesToSend.emplace(id, createTs, datum.size(),
-                                               currMessage.Tx);
+                                               std::move(currMessage.Tx));
             }
         }
         block.Data = std::move(CurrentBatch.Data);
@@ -1529,8 +1538,8 @@ void TWriteSessionImpl::SendImpl() {
                 auto* msgData = writeRequest->add_messages();
 
                 if (message.Tx) {
-                    writeRequest->mutable_tx()->set_id(TStringType{message.Tx->GetId()});
-                    writeRequest->mutable_tx()->set_session(TStringType{message.Tx->GetSession().GetId()});
+                    writeRequest->mutable_tx()->set_id(message.Tx->TxId);
+                    writeRequest->mutable_tx()->set_session(message.Tx->SessionId);
                 }
 
                 msgData->set_seq_no(GetSeqNoImpl(message.Id));

ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.h

@@ -164,18 +164,18 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
         std::optional<ECodec> Codec;
         ui32 OriginalSize; // only for coded messages
         std::vector<std::pair<std::string, std::string>> MessageMeta;
-        const NTable::TTransaction* Tx;
+        std::optional<TTransactionId> Tx;
 
         TMessage(uint64_t id, const TInstant& createdAt, std::string_view data, std::optional<ECodec> codec = {},
                  ui32 originalSize = 0, const std::vector<std::pair<std::string, std::string>>& messageMeta = {},
-                 const NTable::TTransaction* tx = nullptr)
+                 std::optional<TTransactionId>&& tx = {})
             : Id(id)
             , CreatedAt(createdAt)
             , DataRef(data)
             , Codec(codec)
             , OriginalSize(originalSize)
             , MessageMeta(messageMeta)
-            , Tx(tx)
+            , Tx(std::move(tx))
         {}
     };
 
@@ -189,11 +189,11 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
 
         void Add(uint64_t id, const TInstant& createdAt, std::string_view data, std::optional<ECodec> codec, ui32 originalSize,
                  const std::vector<std::pair<std::string, std::string>>& messageMeta,
-                 const NTable::TTransaction* tx) {
+                 std::optional<TTransactionId>&& tx) {
             if (StartedAt == TInstant::Zero())
                 StartedAt = TInstant::Now();
             CurrentSize += codec ? originalSize : data.size();
-            Messages.emplace_back(id, createdAt, data, codec, originalSize, messageMeta, tx);
+            Messages.emplace_back(id, createdAt, data, codec, originalSize, messageMeta, std::move(tx));
             Acquired = false;
         }
 
@@ -263,24 +263,24 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
         TInstant CreatedAt;
         size_t Size;
         std::vector<std::pair<std::string, std::string>> MessageMeta;
-        const NTable::TTransaction* Tx;
+        std::optional<TTransactionId> Tx;
 
         TOriginalMessage(const uint64_t id, const TInstant createdAt, const size_t size,
-                         const NTable::TTransaction* tx)
+                         std::optional<TTransactionId>&& tx)
             : Id(id)
             , CreatedAt(createdAt)
             , Size(size)
-            , Tx(tx)
+            , Tx(std::move(tx))
         {}
 
         TOriginalMessage(const uint64_t id, const TInstant createdAt, const size_t size,
                          std::vector<std::pair<std::string, std::string>>&& messageMeta,
-                         const NTable::TTransaction* tx)
+                         std::optional<TTransactionId>&& tx)
             : Id(id)
             , CreatedAt(createdAt)
             , Size(size)
             , MessageMeta(std::move(messageMeta))
-            , Tx(tx)
+            , Tx(std::move(tx))
         {}
     };