(refactoring) Basic TTopicMessage (ydb-platform#15381)

CyberROFL · nshestakov · commit 7fc930f3abb0 · 2025-05-16T13:08:51.000Z
diff --git a/ydb/core/backup/impl/local_partition_reader.cpp b/ydb/core/backup/impl/local_partition_reader.cpp
@@ -1,12 +1,12 @@
 #include "local_partition_reader.h"
 #include "logging.h"
 
-#include <ydb/library/actors/core/actor.h>
-#include <ydb/library/services/services.pb.h>
-
 #include <ydb/core/persqueue/events/global.h>
 #include <ydb/core/protos/grpc_pq_old.pb.h>
 #include <ydb/core/tx/replication/service/worker.h>
+#include <ydb/core/tx/replication/ydb_proxy/topic_message.h>
+#include <ydb/library/actors/core/actor.h>
+#include <ydb/library/services/services.pb.h>
 
 using namespace NActors;
 using namespace NKikimr::NReplication::NService;
@@ -131,11 +131,11 @@ class TLocalPartitionReader
         }
 
         auto gotOffset = Offset;
-        TVector<TEvWorker::TEvData::TRecord> records(::Reserve(readResult.ResultSize()));
+        TVector<NReplication::TTopicMessage> records(::Reserve(readResult.ResultSize()));
 
         for (auto& result : readResult.GetResult()) {
             gotOffset = std::max(gotOffset, result.GetOffset());
-            records.emplace_back(result.GetOffset(), GetDeserializedData(result.GetData()).GetData(), TInstant::MilliSeconds(result.GetCreateTimestampMS()), result.GetSourceId(), result.GetSourceId(), result.GetSeqNo());
+            records.emplace_back(result.GetOffset(), GetDeserializedData(result.GetData()).GetData());
         }
         SentOffset = gotOffset + 1;
 
diff --git a/ydb/core/backup/impl/local_partition_reader_ut.cpp b/ydb/core/backup/impl/local_partition_reader_ut.cpp
@@ -46,10 +46,10 @@ Y_UNIT_TEST_SUITE(LocalPartitionReader) {
         auto data = runtime.GrabEdgeEventRethrow<TEvWorker::TEvData>(handle);
         UNIT_ASSERT_VALUES_EQUAL(data->Source, PARTITION_STR);
         UNIT_ASSERT_VALUES_EQUAL(data->Records.size(), 2);
-        UNIT_ASSERT_VALUES_EQUAL(data->Records[0].Offset, INITIAL_OFFSET + dataPatternCookie * 2);
-        UNIT_ASSERT_VALUES_EQUAL(data->Records[0].Data, Sprintf("1-%d", dataPatternCookie));
-        UNIT_ASSERT_VALUES_EQUAL(data->Records[1].Offset, INITIAL_OFFSET + dataPatternCookie * 2 + 1);
-        UNIT_ASSERT_VALUES_EQUAL(data->Records[1].Data, Sprintf("2-%d", dataPatternCookie));
+        UNIT_ASSERT_VALUES_EQUAL(data->Records[0].GetOffset(), INITIAL_OFFSET + dataPatternCookie * 2);
+        UNIT_ASSERT_VALUES_EQUAL(data->Records[0].GetData(), Sprintf("1-%d", dataPatternCookie));
+        UNIT_ASSERT_VALUES_EQUAL(data->Records[1].GetOffset(), INITIAL_OFFSET + dataPatternCookie * 2 + 1);
+        UNIT_ASSERT_VALUES_EQUAL(data->Records[1].GetData(), Sprintf("2-%d", dataPatternCookie));
     }
 
     TEvPersQueue::TEvResponse* GenerateData(ui32 dataPatternCookie) {
diff --git a/ydb/core/tx/replication/service/base_table_writer.cpp b/ydb/core/tx/replication/service/base_table_writer.cpp
@@ -8,6 +8,7 @@
 #include <ydb/core/change_exchange/util.h>
 #include <ydb/core/tablet_flat/flat_row_eggs.h>
 #include <ydb/core/tx/datashard/datashard.h>
+#include <ydb/core/tx/replication/ydb_proxy/topic_message.h>
 #include <ydb/core/tx/scheme_cache/helpers.h>
 #include <ydb/core/tx/tx_proxy/proxy.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
@@ -434,8 +435,8 @@ class TLocalTableWriter
         TSet<TRowVersion> versionsWithoutTxId;
 
         for (auto& r : ev->Get()->Records) {
-            auto offset = r.Offset;
-            auto& data = r.Data;
+            auto offset = r.GetOffset();
+            auto& data = r.GetData();
 
             auto record = Parser->Parse(ev->Get()->Source, offset, std::move(data));
 
diff --git a/ydb/core/tx/replication/service/common_ut.h b/ydb/core/tx/replication/service/common_ut.h
@@ -1,12 +1,12 @@
 #pragma once
 
-#include "worker.h"
+#include <ydb/core/tx/replication/ydb_proxy/topic_message.h>
 
 namespace NKikimr::NReplication::NService {
 
-struct TRecord: public TEvWorker::TEvData::TRecord {
+struct TRecord: public TTopicMessage {
     explicit TRecord(ui64 offset, const TString& data)
-        : TEvWorker::TEvData::TRecord(offset, data, TInstant::Zero(), "MessageGroupId", "ProducerId", 42)
+        : TTopicMessage(offset, data)
     {}
 };
 
diff --git a/ydb/core/tx/replication/service/s3_writer_ut.cpp b/ydb/core/tx/replication/service/s3_writer_ut.cpp
@@ -1,3 +1,4 @@
+#include "common_ut.h"
 #include "s3_writer.h"
 #include "worker.h"
 
@@ -60,8 +61,7 @@ Y_UNIT_TEST_SUITE(S3Writer) {
         UNIT_ASSERT_VALUES_EQUAL(s3Mock.GetData().at("/TEST/writer.AtufpxzetsqaVnEuozdXpD.json"),
                                  R"({"finished":false,"table_name":"/MyRoot/Table","writer_name":"AtufpxzetsqaVnEuozdXpD"})");
 
-        using TRecord = TEvWorker::TEvData::TRecord;
-        env.Send<TEvWorker::TEvPoll>(writer, new TEvWorker::TEvData({
+        env.Send<TEvWorker::TEvPoll>(writer, new TEvWorker::TEvData(0, "TestSource", {
             TRecord(1, R"({"key":[1], "update":{"value":"10"}})"),
             TRecord(2, R"({"key":[2], "update":{"value":"20"}})"),
             TRecord(3, R"({"key":[3], "update":{"value":"30"}})"),
@@ -75,7 +75,7 @@ Y_UNIT_TEST_SUITE(S3Writer) {
                                  R"({"key":[2], "update":{"value":"20"}})" "\n"
                                  R"({"key":[3], "update":{"value":"30"}})" "\n");
 
-        auto res = env.Send<TEvWorker::TEvGone>(writer, new TEvWorker::TEvData({}));
+        auto res = env.Send<TEvWorker::TEvGone>(writer, new TEvWorker::TEvData(0, "TestSource", {}));
 
         UNIT_ASSERT_VALUES_EQUAL(res->Get()->Status, TEvWorker::TEvGone::DONE);
         UNIT_ASSERT_VALUES_EQUAL(s3Mock.GetData().size(), 2);
diff --git a/ydb/core/tx/replication/service/table_writer_ut.cpp b/ydb/core/tx/replication/service/table_writer_ut.cpp
@@ -1,6 +1,7 @@
+#include "common_ut.h"
 #include "service.h"
 #include "table_writer.h"
-#include "common_ut.h"
+#include "worker.h"
 
 #include <ydb/core/tx/datashard/ut_common/datashard_ut_common.h>
 #include <ydb/core/tx/replication/ut_helpers/test_env.h>
diff --git a/ydb/core/tx/replication/service/topic_reader.cpp b/ydb/core/tx/replication/service/topic_reader.cpp
@@ -2,6 +2,7 @@
 #include "topic_reader.h"
 #include "worker.h"
 
+#include <ydb/core/tx/replication/ydb_proxy/topic_message.h>
 #include <ydb/core/tx/replication/ydb_proxy/ydb_proxy.h>
 #include <ydb/library/actors/core/actor.h>
 #include <ydb/library/actors/core/hfunc.h>
@@ -57,11 +58,11 @@ class TRemoteTopicReader: public TActor<TRemoteTopicReader> {
         LOG_D("Handle " << ev->Get()->ToString());
 
         auto& result = ev->Get()->Result;
-        TVector<TEvWorker::TEvData::TRecord> records(::Reserve(result.Messages.size()));
+        TVector<TTopicMessage> records(::Reserve(result.Messages.size()));
 
         for (auto& msg : result.Messages) {
             Y_ABORT_UNLESS(msg.GetCodec() == NYdb::NTopic::ECodec::RAW);
-            records.emplace_back(msg.GetOffset(), std::move(msg.GetData()), msg.GetCreateTime(), std::move(msg.GetMessageGroupId()), std::move(msg.GetProducerId()), msg.GetSeqNo());
+            records.push_back(std::move(msg));
         }
 
         Send(Worker, new TEvWorker::TEvData(result.PartitionId, ToString(result.PartitionId), std::move(records)));
diff --git a/ydb/core/tx/replication/service/topic_reader_ut.cpp b/ydb/core/tx/replication/service/topic_reader_ut.cpp
@@ -86,8 +86,8 @@ Y_UNIT_TEST_SUITE(RemoteTopicReader) {
             UNIT_ASSERT_VALUES_EQUAL(records.size(), 1);
 
             const auto& record = records.at(0);
-            UNIT_ASSERT_VALUES_EQUAL(record.Offset, 0);
-            UNIT_ASSERT_VALUES_EQUAL(record.Data, "message-1");
+            UNIT_ASSERT_VALUES_EQUAL(record.GetOffset(), 0);
+            UNIT_ASSERT_VALUES_EQUAL(record.GetData(), "message-1");
         }
 
         // trigger commit, write new data & kill reader
@@ -103,8 +103,8 @@ Y_UNIT_TEST_SUITE(RemoteTopicReader) {
             UNIT_ASSERT_VALUES_EQUAL(records.size(), 1);
 
             const auto& record = records.at(0);
-            UNIT_ASSERT_VALUES_EQUAL(record.Offset, 1);
-            UNIT_ASSERT_VALUES_EQUAL(record.Data, "message-2");
+            UNIT_ASSERT_VALUES_EQUAL(record.GetOffset(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(record.GetData(), "message-2");
         }
     }
 }
diff --git a/ydb/core/tx/replication/service/ut_s3_writer/ya.make b/ydb/core/tx/replication/service/ut_s3_writer/ya.make
@@ -6,6 +6,7 @@ SIZE(MEDIUM)
 
 PEERDIR(
     ydb/core/tx/replication/ut_helpers
+    ydb/core/tx/replication/ydb_proxy
     library/cpp/string_utils/base64
     library/cpp/testing/unittest
 )
diff --git a/ydb/core/tx/replication/service/ut_table_writer/ya.make b/ydb/core/tx/replication/service/ut_table_writer/ya.make
@@ -7,6 +7,7 @@ SIZE(MEDIUM)
 PEERDIR(
     ydb/core/tx/datashard/ut_common
     ydb/core/tx/replication/ut_helpers
+    ydb/core/tx/replication/ydb_proxy
     library/cpp/string_utils/base64
     library/cpp/testing/unittest
 )
diff --git a/ydb/core/tx/replication/service/ut_transfer_writer/ya.make b/ydb/core/tx/replication/service/ut_transfer_writer/ya.make
@@ -7,6 +7,7 @@ SIZE(MEDIUM)
 PEERDIR(
     ydb/core/tx/datashard/ut_common
     ydb/core/tx/replication/ut_helpers
+    ydb/core/tx/replication/ydb_proxy
     library/cpp/string_utils/base64
     library/cpp/testing/unittest
 )
diff --git a/ydb/core/tx/replication/service/worker.cpp b/ydb/core/tx/replication/service/worker.cpp
@@ -3,6 +3,7 @@
 #include "worker.h"
 
 #include <ydb/core/base/appdata.h>
+#include <ydb/core/tx/replication/ydb_proxy/topic_message.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
 #include <ydb/library/actors/core/hfunc.h>
 #include <ydb/library/services/services.pb.h>
@@ -13,26 +14,6 @@
 
 namespace NKikimr::NReplication::NService {
 
-TEvWorker::TEvData::TRecord::TRecord(ui64 offset, const TString& data, TInstant createTime, const TString& messageGroupId, const TString& producerId, ui64 seqNo)
-    : Offset(offset)
-    , Data(data)
-    , CreateTime(createTime)
-    , MessageGroupId(messageGroupId)
-    , ProducerId(producerId)
-    , SeqNo(seqNo)
-{
-}
-
-TEvWorker::TEvData::TRecord::TRecord(ui64 offset, TString&& data, TInstant createTime, TString&& messageGroupId, TString&& producerId, ui64 seqNo)
-    : Offset(offset)
-    , Data(std::move(data))
-    , CreateTime(createTime)
-    , MessageGroupId(std::move(messageGroupId))
-    , ProducerId(std::move(producerId))
-    , SeqNo(seqNo)
-{
-}
-
 TEvWorker::TEvPoll::TEvPoll(bool skipCommit)
     : SkipCommit(skipCommit)
 {
@@ -55,28 +36,20 @@ TString TEvWorker::TEvCommit::ToString() const {
     << " }";
 }
 
-TEvWorker::TEvData::TEvData(ui32 partitionId, const TString& source, const TVector<TRecord>& records)
+TEvWorker::TEvData::TEvData(ui32 partitionId, const TString& source, const TVector<TTopicMessage>& records)
     : PartitionId(partitionId)
     , Source(source)
     , Records(records)
 {
 }
 
-TEvWorker::TEvData::TEvData(ui32 partitionId, const TString& source, TVector<TRecord>&& records)
+TEvWorker::TEvData::TEvData(ui32 partitionId, const TString& source, TVector<TTopicMessage>&& records)
     : PartitionId(partitionId)
     , Source(source)
     , Records(std::move(records))
 {
 }
 
-void TEvWorker::TEvData::TRecord::Out(IOutputStream& out) const {
-    out << "{"
-        << " Offset: " << Offset
-        << " Data: " << Data.size() << "b"
-        << " CreateTime: " << CreateTime.ToStringUpToSeconds()
-    << " }";
-}
-
 TString TEvWorker::TEvData::ToString() const {
     return TStringBuilder() << ToStringHeader() << " {"
         << " Source: " << Source
@@ -211,11 +184,11 @@ class TWorker: public TActorBootstrapped<TWorker> {
         if (InFlightData) {
             const auto& records = InFlightData->Records;
             auto it = MinElementBy(records, [](const auto& record) {
-                return record.CreateTime;
+                return record.GetCreateTime();
             });
 
             if (it != records.end()) {
-                Lag = TlsActivationContext->Now() - it->CreateTime;
+                Lag = TlsActivationContext->Now() - it->GetCreateTime();
             }
         }
 
diff --git a/ydb/core/tx/replication/service/worker.h b/ydb/core/tx/replication/service/worker.h
@@ -8,7 +8,11 @@
 
 #include <functional>
 
-namespace NKikimr::NReplication::NService {
+namespace NKikimr::NReplication {
+
+class TTopicMessage;
+
+namespace NService {
 
 struct TEvWorker {
     enum EEv {
@@ -44,25 +48,12 @@ struct TEvWorker {
     };
 
     struct TEvData: public TEventLocal<TEvData, EvData> {
-        struct TRecord {
-            ui64 Offset;
-            TString Data;
-            TInstant CreateTime;
-            TString MessageGroupId;
-            TString ProducerId;
-            ui64 SeqNo;
-
-            explicit TRecord(ui64 offset, const TString& data, TInstant createTime, const TString& messageGroupId, const TString& producerId, ui64 seqNo);
-            explicit TRecord(ui64 offset, TString&& data, TInstant createTime, TString&& messageGroupId, TString&& producerId, ui64 seqNo);
-            void Out(IOutputStream& out) const;
-        };
-
         ui32 PartitionId;
         TString Source;
-        TVector<TRecord> Records;
+        TVector<TTopicMessage> Records;
 
-        explicit TEvData(ui32 partitionId, const TString& source, const TVector<TRecord>& records);
-        explicit TEvData(ui32 partitionId, const TString& source, TVector<TRecord>&& records);
+        explicit TEvData(ui32 partitionId, const TString& source, const TVector<TTopicMessage>& records);
+        explicit TEvData(ui32 partitionId, const TString& source, TVector<TTopicMessage>&& records);
         TString ToString() const override;
     };
 
@@ -103,8 +94,5 @@ IActor* CreateWorker(
     std::function<IActor*(void)>&& createReaderFn,
     std::function<IActor*(void)>&& createWriterFn);
 
-}
-
-Y_DECLARE_OUT_SPEC(inline, NKikimr::NReplication::NService::TEvWorker::TEvData::TRecord, o, x) {
-    return x.Out(o);
-}
+} // NService
+} // NKikimr::NReplication
diff --git a/ydb/core/tx/replication/ydb_proxy/topic_message.cpp b/ydb/core/tx/replication/ydb_proxy/topic_message.cpp
diff --git a/ydb/core/tx/replication/ydb_proxy/topic_message.h b/ydb/core/tx/replication/ydb_proxy/topic_message.h
diff --git a/ydb/core/tx/replication/ydb_proxy/ya.make b/ydb/core/tx/replication/ydb_proxy/ya.make
diff --git a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp
diff --git a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h

Original file line number	Diff line number	Diff line change
`@@ -86,8 +86,8 @@ Y_UNIT_TEST_SUITE(RemoteTopicReader) {`
`86`	`86`	`UNIT_ASSERT_VALUES_EQUAL(records.size(), 1);`
`87`	`87`
`88`	`88`	`const auto& record = records.at(0);`
`89`		`- UNIT_ASSERT_VALUES_EQUAL(record.Offset, 0);`
`90`		`- UNIT_ASSERT_VALUES_EQUAL(record.Data, "message-1");`
	`89`	`+ UNIT_ASSERT_VALUES_EQUAL(record.GetOffset(), 0);`
	`90`	`+ UNIT_ASSERT_VALUES_EQUAL(record.GetData(), "message-1");`
`91`	`91`	`}`
`92`	`92`
`93`	`93`	`// trigger commit, write new data & kill reader`
`@@ -103,8 +103,8 @@ Y_UNIT_TEST_SUITE(RemoteTopicReader) {`
`103`	`103`	`UNIT_ASSERT_VALUES_EQUAL(records.size(), 1);`
`104`	`104`
`105`	`105`	`const auto& record = records.at(0);`
`106`		`- UNIT_ASSERT_VALUES_EQUAL(record.Offset, 1);`
`107`		`- UNIT_ASSERT_VALUES_EQUAL(record.Data, "message-2");`
	`106`	`+ UNIT_ASSERT_VALUES_EQUAL(record.GetOffset(), 1);`
	`107`	`+ UNIT_ASSERT_VALUES_EQUAL(record.GetData(), "message-2");`
`108`	`108`	`}`
`109`	`109`	`}`
`110`	`110`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@ SIZE(MEDIUM)`
`6`	`6`
`7`	`7`	`PEERDIR(`
`8`	`8`	`ydb/core/tx/replication/ut_helpers`
	`9`	`+ ydb/core/tx/replication/ydb_proxy`
`9`	`10`	`library/cpp/string_utils/base64`
`10`	`11`	`library/cpp/testing/unittest`
`11`	`12`	`)`
Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@ SIZE(MEDIUM)`
`7`	`7`	`PEERDIR(`
`8`	`8`	`ydb/core/tx/datashard/ut_common`
`9`	`9`	`ydb/core/tx/replication/ut_helpers`
	`10`	`+ ydb/core/tx/replication/ydb_proxy`
`10`	`11`	`library/cpp/string_utils/base64`
`11`	`12`	`library/cpp/testing/unittest`
`12`	`13`	`)`