Move transfer to internal repository (#16671)

Author: nshestakov

ydb/core/base/appdata_fwd.h

@@ -153,6 +153,10 @@ namespace NSchemeShard {
     class IOperationFactory;
 }
 
+namespace NReplication::NService {
+    class ITransferWriterFactory;
+}
+
 class TFormatFactory;
 
 namespace NYamlConfig {
@@ -188,6 +192,7 @@ struct TAppData {
 
     const NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory* PersQueueGetReadSessionsInfoWorkerFactory = nullptr;
     const NPQ::IPersQueueMirrorReaderFactory* PersQueueMirrorReaderFactory = nullptr;
+    std::shared_ptr<NReplication::NService::ITransferWriterFactory> TransferWriterFactory = nullptr;
     NYdb::TDriver* YdbDriver = nullptr;
     const NPDisk::IIoContextFactory* IoContextFactory = nullptr;
 

ydb/core/driver_lib/run/factories.h

@@ -2,6 +2,7 @@
 #include <ydb/core/blobstorage/pdisk/blobstorage_pdisk_util_devicemode.h>
 #include <ydb/core/kqp/common/kqp.h>
 #include <ydb/core/tx/datashard/export_iface.h>
+#include <ydb/core/tx/replication/service/transfer_writer_factory.h>
 #include <ydb/core/tx/schemeshard/schemeshard_operation_factory.h>
 #include <ydb/core/persqueue/actor_persqueue_client_iface.h>
 #include <ydb/core/protos/auth.pb.h>
@@ -53,6 +54,7 @@ struct TModuleFactories {
     TGrpcServiceFactory GrpcServiceFactory;
 
     std::shared_ptr<NPQ::IPersQueueMirrorReaderFactory> PersQueueMirrorReaderFactory;
+    std::shared_ptr<NReplication::NService::ITransferWriterFactory> TransferWriterFactory;
     /// Factory for pdisk's aio engines
     std::shared_ptr<NPDisk::IIoContextFactory> IoContextFactory;
 

ydb/core/driver_lib/run/run.cpp

@@ -1127,6 +1127,9 @@ void TKikimrRunner::InitializeAppData(const TKikimrRunConfig& runConfig)
     AppData->IoContextFactory = ModuleFactories ? ModuleFactories->IoContextFactory.get() : nullptr;
     AppData->SchemeOperationFactory = ModuleFactories ? ModuleFactories->SchemeOperationFactory.get() : nullptr;
     AppData->ConfigSwissKnife = ModuleFactories ? ModuleFactories->ConfigSwissKnife.get() : nullptr;
+    if (ModuleFactories) {
+        AppData->TransferWriterFactory = ModuleFactories->TransferWriterFactory;
+    }
 
     AppData->SqsAuthFactory = ModuleFactories
         ? ModuleFactories->SqsAuthFactory.get()

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -26,6 +26,7 @@ namespace NKqp {
 
 using namespace NYdb;
 using namespace NYdb::NTable;
+using namespace NYdb::NReplication;
 
 Y_UNIT_TEST_SUITE(KqpScheme) {
     Y_UNIT_TEST(UseUnauthorizedTable) {

ydb/core/testlib/actors/test_runtime.cpp

@@ -8,6 +8,7 @@
 #include <ydb/core/mon_alloc/profiler.h>
 #include <ydb/core/grpc_services/grpc_helper.h>
 #include <ydb/core/tablet/tablet_impl.h>
+#include <ydb/core/testlib/mock_transfer_writer_factory.h>
 
 #include <ydb/library/actors/core/executor_pool_basic.h>
 #include <ydb/library/actors/core/executor_pool_io.h>
@@ -182,6 +183,7 @@ namespace NActors {
             nodeAppData->EnableMvccSnapshotWithLegacyDomainRoot = app0->EnableMvccSnapshotWithLegacyDomainRoot;
             nodeAppData->IoContextFactory = app0->IoContextFactory;
             nodeAppData->SchemeOperationFactory = app0->SchemeOperationFactory;
+            nodeAppData->TransferWriterFactory = std::make_shared<NKikimr::Tests::MockTransferWriterFactory>();
             if (nodeIndex < egg.Icb.size()) {
                 nodeAppData->Icb = std::move(egg.Icb[nodeIndex]);
                 nodeAppData->InFlightLimiterRegistry.Reset(new NKikimr::NGRpcService::TInFlightLimiterRegistry(nodeAppData->Icb));

ydb/core/testlib/mock_transfer_writer_factory.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include <ydb/core/tx/replication/service/transfer_writer_factory.h>
+#include <ydb/library/actors/core/actor_bootstrapped.h>
+
+namespace NKikimr::Tests {
+
+struct MockTransferWriterFactory : public NKikimr::NReplication::NService::ITransferWriterFactory {
+    struct MockActor : public NActors::TActorBootstrapped<MockActor> {
+        void Bootstrap() {}
+    };
+
+
+    NActors::IActor* Create(const Parameters&) const override {
+        return new MockActor();
+    }
+};
+
+} // namespace NKikimr::Tests

ydb/core/testlib/test_client.cpp

@@ -538,6 +538,7 @@ namespace Tests {
             appData.GraphConfig.MergeFrom(Settings->AppConfig->GetGraphConfig());
             appData.SqsConfig.MergeFrom(Settings->AppConfig->GetSqsConfig());
             appData.SharedCacheConfig.MergeFrom(Settings->AppConfig->GetSharedCacheConfig());
+            appData.TransferWriterFactory = Settings->TransferWriterFactory;
 
             appData.DynamicNameserviceConfig = new TDynamicNameserviceConfig;
             auto dnConfig = appData.DynamicNameserviceConfig;

ydb/core/testlib/test_client.h

@@ -24,6 +24,7 @@
 #include <ydb/core/protos/flat_scheme_op.pb.h>
 #include <ydb/core/testlib/basics/runtime.h>
 #include <ydb/core/testlib/basics/appdata.h>
+#include <ydb/core/testlib/mock_transfer_writer_factory.h>
 #include <ydb/core/protos/kesus.pb.h>
 #include <ydb/core/protos/table_service_config.pb.h>
 #include <ydb/core/protos/console_tenant.pb.h>
@@ -157,6 +158,8 @@ namespace Tests {
         std::shared_ptr<NKikimr::NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> PersQueueGetReadSessionsInfoWorkerFactory;
         std::shared_ptr<NKikimr::NHttpProxy::IAuthFactory> DataStreamsAuthFactory;
         std::shared_ptr<NKikimr::NPQ::TPersQueueMirrorReaderFactory> PersQueueMirrorReaderFactory = std::make_shared<NKikimr::NPQ::TPersQueueMirrorReaderFactory>();
+        std::shared_ptr<NKikimr::NReplication::NService::ITransferWriterFactory> TransferWriterFactory = std::make_shared<MockTransferWriterFactory>();
+
         bool EnableMetering = false;
         TString MeteringFilePath;
         TString AwsRegion;

ydb/core/tx/replication/service/service.cpp

@@ -2,7 +2,7 @@
 #include "service.h"
 #include "table_writer.h"
 #include "topic_reader.h"
-#include "transfer_writer.h"
+#include "transfer_writer_factory.h"
 #include "worker.h"
 
 #include <ydb/core/base/appdata.h>
@@ -417,7 +417,8 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
     }
 
     std::function<IActor*(void)> TransferWriterFn(
-            const NKikimrReplication::TTransferWriterSettings& writerSettings)
+            const NKikimrReplication::TTransferWriterSettings& writerSettings,
+            const ITransferWriterFactory* transferWriterFactory)
     {
         if (!CompilationService) {
             CompilationService = Register(
@@ -429,9 +430,10 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
             tablePathId = TPathId::FromProto(writerSettings.GetPathId()),
             transformLambda = writerSettings.GetTransformLambda(),
             compilationService = *CompilationService,
-            batchingSettings = writerSettings.GetBatching()
+            batchingSettings = writerSettings.GetBatching(),
+            transferWriterFactory = transferWriterFactory
         ]() {
-            return CreateTransferWriter(transformLambda, tablePathId, compilationService, batchingSettings);
+            return transferWriterFactory->Create({transformLambda, tablePathId, compilationService, batchingSettings});
         };
     }
 
@@ -478,7 +480,12 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
             writerFn = WriterFn(writerSettings, consistencySettings);
         } else if (cmd.HasTransferWriter()) {
             const auto& writerSettings = cmd.GetTransferWriter();
-            writerFn = TransferWriterFn(writerSettings);
+            const auto* transferWriterFactory = AppData()->TransferWriterFactory.get();
+            if (!transferWriterFactory) {
+                LOG_C("Run transfer but TransferWriterFactory does not exists.");
+                return;
+            }
+            writerFn = TransferWriterFn(writerSettings, transferWriterFactory);
         } else {
             Y_ABORT("Unsupported");
         }