Fixed committing of proccessed messages by transfer (#17580)

nshestakov · nshestakov · commit 8c3f4c162cb0 · 2025-05-05T08:55:50.000Z
diff --git a/ydb/core/tx/replication/service/service.cpp b/ydb/core/tx/replication/service/service.cpp
@@ -377,7 +377,7 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
         return it->second;
     }
 
-    std::function<IActor*(void)> ReaderFn(const NKikimrReplication::TRemoteTopicReaderSettings& settings) {
+    std::function<IActor*(void)> ReaderFn(const NKikimrReplication::TRemoteTopicReaderSettings& settings, bool autoCommit) {
         TActorId ydbProxy;
         const auto& params = settings.GetConnectionParams();
         switch (params.GetCredentialsCase()) {
@@ -394,6 +394,7 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
         auto topicReaderSettings = TEvYdbProxy::TTopicReaderSettings()
             .MaxMemoryUsageBytes(1_MB)
             .ConsumerName(settings.GetConsumerName())
+            .AutoCommit(autoCommit)
             .AppendTopics(NYdb::NTopic::TTopicReadSettings()
                 .Path(settings.GetTopicPath())
                 .AppendPartitionIds(settings.GetTopicPartitionId())
@@ -473,6 +474,7 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
         const auto& cmd = record.GetCommand();
         // TODO: validate settings
         const auto& readerSettings = cmd.GetRemoteTopicReader();
+        bool autoCommit = true;
         std::function<IActor*(void)> writerFn;
         if (cmd.HasLocalTableWriter()) {
             const auto& writerSettings = cmd.GetLocalTableWriter();
@@ -485,12 +487,13 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
                 LOG_C("Run transfer but TransferWriterFactory does not exists.");
                 return;
             }
+            autoCommit = false;
             writerFn = TransferWriterFn(writerSettings, transferWriterFactory);
         } else {
             Y_ABORT("Unsupported");
         }
         const auto actorId = session.RegisterWorker(this, id,
-            CreateWorker(SelfId(), ReaderFn(readerSettings), std::move(writerFn)));
+            CreateWorker(SelfId(), ReaderFn(readerSettings, autoCommit), std::move(writerFn)));
         WorkerActorIdToSession[actorId] = controller.GetTabletId();
     }
 
diff --git a/ydb/core/tx/replication/service/topic_reader.cpp b/ydb/core/tx/replication/service/topic_reader.cpp
@@ -67,13 +67,44 @@ class TRemoteTopicReader: public TActor<TRemoteTopicReader> {
         Send(Worker, new TEvWorker::TEvData(result.PartitionId, ToString(result.PartitionId), std::move(records)));
     }
 
-    void Handle(TEvYdbProxy::TEvTopicEndPartition::TPtr& ev) {
+    void Handle(TEvYdbProxy::TEvEndTopicPartition::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
 
         auto& result = ev->Get()->Result;
         Send(Worker, new TEvWorker::TEvDataEnd(result.PartitionId, std::move(result.AdjacentPartitionsIds), std::move(result.ChildPartitionsIds)));
     }
 
+    void Handle(TEvYdbProxy::TEvStartTopicReadingSession::TPtr& ev) {
+        LOG_D("Handle " << ev->Get()->ToString());
+
+        ReadSessionId = ev->Get()->Result.ReadSessionId;
+    }
+
+    void Handle(TEvWorker::TEvCommit::TPtr& ev) {
+        LOG_D("Handle " << ev->Get()->ToString());
+
+        Y_ABORT_UNLESS(YdbProxy);
+        Y_ABORT_UNLESS(ReadSessionId);
+
+        auto settings = NYdb::NTopic::TCommitOffsetSettings()
+            .ReadSessionId(ReadSessionId);
+
+        const auto& topicName = Settings.GetBase().Topics_.at(0).Path_;
+        const auto partitionId = Settings.GetBase().Topics_.at(0).PartitionIds_.at(0);
+        const auto& consumerName = Settings.GetBase().ConsumerName_;
+
+        Send(YdbProxy, new TEvYdbProxy::TEvCommitOffsetRequest(topicName, partitionId, consumerName, ev->Get()->Offset, std::move(settings)));
+    }
+
+    void Handle(TEvYdbProxy::TEvCommitOffsetResponse::TPtr& ev) {
+        if (!ev->Get()->Result.IsSuccess()) {
+            LOG_W("Handle " << ev->Get()->ToString());
+            return Leave(TEvWorker::TEvGone::UNAVAILABLE);
+        } else {
+            LOG_D("Handle " << ev->Get()->ToString());
+        }
+    }
+
     void Handle(TEvYdbProxy::TEvTopicReaderGone::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
 
@@ -120,9 +151,12 @@ class TRemoteTopicReader: public TActor<TRemoteTopicReader> {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvWorker::TEvHandshake, Handle);
             hFunc(TEvWorker::TEvPoll, Handle);
+            hFunc(TEvWorker::TEvCommit, Handle);
             hFunc(TEvYdbProxy::TEvCreateTopicReaderResponse, Handle);
             hFunc(TEvYdbProxy::TEvReadTopicResponse, Handle);
-            hFunc(TEvYdbProxy::TEvTopicEndPartition, Handle);
+            hFunc(TEvYdbProxy::TEvCommitOffsetResponse, Handle);
+            hFunc(TEvYdbProxy::TEvStartTopicReadingSession, Handle);
+            hFunc(TEvYdbProxy::TEvEndTopicPartition, Handle);
             hFunc(TEvYdbProxy::TEvTopicReaderGone, Handle);
             sFunc(TEvents::TEvPoison, PassAway);
         }
@@ -135,6 +169,7 @@ class TRemoteTopicReader: public TActor<TRemoteTopicReader> {
 
     TActorId Worker;
     TActorId ReadSession;
+    TString ReadSessionId;
 
 }; // TRemoteTopicReader
 
diff --git a/ydb/core/tx/replication/service/worker.cpp b/ydb/core/tx/replication/service/worker.cpp
@@ -44,6 +44,17 @@ TString TEvWorker::TEvPoll::ToString() const {
     << " }";
 }
 
+TEvWorker::TEvCommit::TEvCommit(size_t offset)
+    : Offset(offset)
+{
+}
+
+TString TEvWorker::TEvCommit::ToString() const {
+    return TStringBuilder() << ToStringHeader() << " {"
+        << " Offset: " << Offset
+    << " }";
+}
+
 TEvWorker::TEvData::TEvData(ui32 partitionId, const TString& source, const TVector<TRecord>& records)
     : PartitionId(partitionId)
     , Source(source)
@@ -214,6 +225,20 @@ class TWorker: public TActorBootstrapped<TWorker> {
         }
     }
 
+    void Handle(TEvWorker::TEvCommit::TPtr& ev) {
+        LOG_D("Handle " << ev->Get()->ToString());
+
+        if (ev->Sender != Writer) {
+            LOG_W("Commit from unknown actor"
+                << ": sender# " << ev->Sender);
+            return;
+        }
+
+        if (Reader) {
+            Send(ev->Forward(Reader));
+        }
+    }
+
     void Handle(TEvWorker::TEvData::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
 
@@ -335,6 +360,7 @@ class TWorker: public TActorBootstrapped<TWorker> {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvWorker::TEvHandshake, Handle);
             hFunc(TEvWorker::TEvPoll, Handle);
+            hFunc(TEvWorker::TEvCommit, Handle);
             hFunc(TEvWorker::TEvData, Handle);
             hFunc(TEvWorker::TEvDataEnd, Forward);
             hFunc(TEvWorker::TEvGone, Handle);
diff --git a/ydb/core/tx/replication/service/worker.h b/ydb/core/tx/replication/service/worker.h
@@ -20,6 +20,7 @@ struct TEvWorker {
         EvGone,
         EvStatus,
         EvDataEnd,
+        EvCommit,
 
         EvEnd,
     };
@@ -35,6 +36,13 @@ struct TEvWorker {
         TString ToString() const override;
     };
 
+    struct TEvCommit: public TEventLocal<TEvCommit, EvCommit> {
+        size_t Offset;
+
+        explicit TEvCommit(size_t offset);
+        TString ToString() const override;
+    };
+
     struct TEvData: public TEventLocal<TEvData, EvData> {
         struct TRecord {
             ui64 Offset;
diff --git a/ydb/core/tx/replication/ydb_proxy/partition_end_watcher.h b/ydb/core/tx/replication/ydb_proxy/partition_end_watcher.h
@@ -14,7 +14,7 @@ class TPartitionEndWatcher {
             return;
         }
 
-        ActorOps->Send(client, new TEvYdbProxy::TEvTopicEndPartition(*EndPartitionSessionEvent));
+        ActorOps->Send(client, new TEvYdbProxy::TEvEndTopicPartition(*EndPartitionSessionEvent));
     }
 
 public:
diff --git a/ydb/core/tx/replication/ydb_proxy/partition_end_watcher_ut.cpp b/ydb/core/tx/replication/ydb_proxy/partition_end_watcher_ut.cpp
@@ -94,7 +94,7 @@ Y_UNIT_TEST_SUITE(PartitionEndWatcher) {
 
         UNIT_ASSERT_VALUES_EQUAL(actorOps.Events.size(), 1);
         UNIT_ASSERT_VALUES_EQUAL(actorOps.Events[0].first, client);
-        auto* e = dynamic_cast<TEvYdbProxy::TEvTopicEndPartition*>(actorOps.Events[0].second);
+        auto* e = dynamic_cast<TEvYdbProxy::TEvEndTopicPartition*>(actorOps.Events[0].second);
         UNIT_ASSERT(e);
         UNIT_ASSERT_VALUES_EQUAL(e->Result.AdjacentPartitionsIds, TVector<ui64>{1});
         UNIT_ASSERT_VALUES_EQUAL(e->Result.ChildPartitionsIds, TVector<ui64>{2});
@@ -118,7 +118,7 @@ Y_UNIT_TEST_SUITE(PartitionEndWatcher) {
 
         UNIT_ASSERT_VALUES_EQUAL(actorOps.Events.size(), 1);
         UNIT_ASSERT_VALUES_EQUAL(actorOps.Events[0].first, client);
-        auto* e = dynamic_cast<TEvYdbProxy::TEvTopicEndPartition*>(actorOps.Events[0].second);
+        auto* e = dynamic_cast<TEvYdbProxy::TEvEndTopicPartition*>(actorOps.Events[0].second);
         UNIT_ASSERT(e);
         UNIT_ASSERT_VALUES_EQUAL(e->Result.AdjacentPartitionsIds, TVector<ui64>{1});
         UNIT_ASSERT_VALUES_EQUAL(e->Result.ChildPartitionsIds, TVector<ui64>{2});
diff --git a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp
@@ -47,6 +47,12 @@ void TEvYdbProxy::TEndTopicPartitionResult::Out(IOutputStream& out) const {
     << " }";
 }
 
+void TEvYdbProxy::TStartTopicReadingSessionResult::Out(IOutputStream& out) const {
+    out << "{"
+        << " ReadSessionId: " << ReadSessionId
+    << " }";
+}
+
 template <typename TDerived>
 class TBaseProxyActor: public TActor<TDerived> {
     class TRequest;
@@ -213,6 +219,7 @@ class TTopicReader: public TBaseProxyActor<TTopicReader> {
         if (auto* x = std::get_if<TReadSessionEvent::TStartPartitionSessionEvent>(&*event)) {
             PartitionEndWatcher.Clear();
             x->Confirm();
+            Send(ev->Get()->Sender, new TEvYdbProxy::TEvStartTopicReadingSession(*x), 0, ev->Get()->Cookie);
             return WaitEvent(ev->Get()->Sender, ev->Get()->Cookie);
         } else if (auto* x = std::get_if<TReadSessionEvent::TStopPartitionSessionEvent>(&*event)) {
             x->Confirm();
diff --git a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h
@@ -51,7 +51,8 @@ struct TEvYdbProxy {
         EvTopicReaderGone,
         EV_REQUEST_RESPONSE(ReadTopic),
         EV_REQUEST_RESPONSE(CommitOffset),
-        EvTopicEndPartition,
+        EvEndTopicPartition,
+        EvStartTopicReadingSession,
 
         EvEnd,
     };
@@ -246,7 +247,22 @@ struct TEvYdbProxy {
         TVector<ui64> ChildPartitionsIds;
     };
 
-    struct TEvTopicEndPartition: public TGenericResponse<TEvTopicEndPartition, EvTopicEndPartition, TEndTopicPartitionResult> {
+    struct TEvEndTopicPartition: public TGenericResponse<TEvEndTopicPartition, EvEndTopicPartition, TEndTopicPartitionResult> {
+        using TBase::TBase;
+    };
+
+    struct TStartTopicReadingSessionResult {
+        explicit TStartTopicReadingSessionResult(const NYdb::NTopic::TReadSessionEvent::TStartPartitionSessionEvent& event)
+            : ReadSessionId(event.GetPartitionSession()->GetReadSessionId())
+        {
+        }
+
+        void Out(IOutputStream& out) const;
+
+        TString ReadSessionId;
+    };
+
+    struct TEvStartTopicReadingSession: public TGenericResponse<TEvStartTopicReadingSession, EvStartTopicReadingSession, TStartTopicReadingSessionResult> {
         using TBase::TBase;
     };
 
@@ -287,7 +303,7 @@ struct TEvYdbProxy {
     DEFINE_GENERIC_REQUEST_RESPONSE(DescribeConsumer, NYdb::NTopic::TDescribeConsumerResult, TString, TString, NYdb::NTopic::TDescribeConsumerSettings);
     DEFINE_GENERIC_REQUEST_RESPONSE(CreateTopicReader, TActorId, TTopicReaderSettings);
     DEFINE_GENERIC_REQUEST_RESPONSE(ReadTopic, TReadTopicResult, TReadTopicSettings);
-    DEFINE_GENERIC_REQUEST_RESPONSE(CommitOffset, NYdb::TStatus, TString, ui64, TString, ui64, NYdb::NTopic::TCommitOffsetSettings);
+    DEFINE_GENERIC_REQUEST_RESPONSE(CommitOffset, NYdb::TStatus, std::string, ui64, std::string, ui64, NYdb::NTopic::TCommitOffsetSettings);
 
     #undef DEFINE_GENERIC_REQUEST_RESPONSE
     #undef DEFINE_GENERIC_RESPONSE
diff --git a/ydb/tests/functional/replication/utils.h b/ydb/tests/functional/replication/utils.h

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,7 @@ class TPartitionEndWatcher {`
`14`	`14`	`return;`
`15`	`15`	`}`
`16`	`16`
`17`		`- ActorOps->Send(client, new TEvYdbProxy::TEvTopicEndPartition(*EndPartitionSessionEvent));`
	`17`	`+ ActorOps->Send(client, new TEvYdbProxy::TEvEndTopicPartition(*EndPartitionSessionEvent));`
`18`	`18`	`}`
`19`	`19`
`20`	`20`	`public:`