fix

Author: VPolka

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

@@ -5870,6 +5870,16 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             const auto result = session.ExecuteSchemeQuery(query).GetValueSync();
             UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
 
+            {
+                Sleep(TDuration::Seconds(10));
+
+                const auto result = repl.DescribeReplication("/Root/replication").ExtractValueSync();
+                UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+                const auto& desc = result.GetReplicationDescription();
+                UNIT_ASSERT_VALUES_EQUAL(desc.GetState(), TReplicationDescription::EState::Done);
+            }
+
             while (true) {
                 const auto result = repl.DescribeReplication("/Root/replication").ExtractValueSync();
                 UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());

ydb/core/tx/replication/controller/logging.h

@@ -3,11 +3,11 @@
 #include <ydb/library/actors/core/log.h>
 #include <ydb/library/services/services.pb.h>
 
-#define CLOG_T(ctx, stream) LOG_TRACE_S (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
-#define CLOG_D(ctx, stream) LOG_DEBUG_S (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
-#define CLOG_I(ctx, stream) LOG_INFO_S  (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
-#define CLOG_N(ctx, stream) LOG_NOTICE_S(ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
-#define CLOG_W(ctx, stream) LOG_WARN_S  (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
+#define CLOG_T(ctx, stream) LOG_ERROR_S (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
+#define CLOG_D(ctx, stream) LOG_ERROR_S (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
+#define CLOG_I(ctx, stream) LOG_ERROR_S  (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
+#define CLOG_N(ctx, stream) LOG_ERROR_S(ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
+#define CLOG_W(ctx, stream) LOG_ERROR_S  (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
 #define CLOG_E(ctx, stream) LOG_ERROR_S (ctx, NKikimrServices::REPLICATION_CONTROLLER, LogPrefix << stream)
 
 #define LOG_T(stream) CLOG_T(*TlsActivationContext, stream)

ydb/core/tx/replication/controller/replication.h

@@ -87,7 +87,7 @@ class TReplication: public TSimpleRefCount<TReplication> {
 
     protected:
         virtual IActor* CreateWorkerRegistar(const TActorContext& ctx) const = 0;
-        virtual IActor* CreateWorkerStoper(const TActorContext& ctx) const = 0;
+        virtual IActor* CreateWorkerStopper(const TActorContext& ctx) const = 0;
     };
 
     friend class TTargetBase;

ydb/core/tx/replication/controller/target_base.cpp

@@ -133,8 +133,8 @@ void TTargetBase::Progress(const TActorContext& ctx) {
         }
         break;
     case EDstState::Pausing:
-        if (!WorkerStoper) {
-            WorkerStoper = ctx.Register(CreateWorkerStoper(ctx));
+        if (!WorkerStopper) {
+            WorkerStopper = ctx.Register(CreateWorkerStopper(ctx));
         }
         break;
     case EDstState::Paused:
@@ -158,7 +158,7 @@ void TTargetBase::Shutdown(const TActorContext& ctx) {
         &DstAlterer,
         &DstRemover,
         &WorkerRegistar,
-        &WorkerStoper,
+        &WorkerStopper,
     };
 
     for (auto* x : toShutdown) {

ydb/core/tx/replication/controller/target_base.h

@@ -65,7 +65,7 @@ class TTargetBase: public TReplication::ITarget {
     TActorId DstAlterer;
     TActorId DstRemover;
     TActorId WorkerRegistar;
-    TActorId WorkerStoper;
+    TActorId WorkerStopper;
     THashSet<ui64> Workers;
     bool PendingRemoveWorkers = false;
 

ydb/core/tx/replication/controller/target_table.cpp

@@ -103,7 +103,7 @@ class TTableWorkerRegistar: public TActorBootstrapped<TTableWorkerRegistar> {
 
 }; // TTableWorkerRegistar
 
-class TTableWorkerStoper: public TActorBootstrapped<TTableWorkerStoper> {
+class TTableWorkerStopper: public TActorBootstrapped<TTableWorkerStopper> {
     void Handle(TEvYdbProxy::TEvDescribeTopicResponse::TPtr& ev) {
         LOG_T("Handle " << ev->Get()->ToString());
 
@@ -139,10 +139,10 @@ class TTableWorkerStoper: public TActorBootstrapped<TTableWorkerStoper> {
 
 public:
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
-        return NKikimrServices::TActivity::REPLICATION_CONTROLLER_TABLE_WORKER_STOPER;
+        return NKikimrServices::TActivity::REPLICATION_CONTROLLER_TABLE_WORKER_STOPPER;
     }
 
-    explicit TTableWorkerStoper(
+    explicit TTableWorkerStopper(
             const TActorId& parent,
             const TActorId& proxy,
             const NKikimrReplication::TConnectionParams& connectionParams,
@@ -157,7 +157,7 @@ class TTableWorkerStoper: public TActorBootstrapped<TTableWorkerStoper> {
         , TargetId(tid)
         , SrcStreamPath(srcStreamPath)
         , DstPathId(dstPathId)
-        , LogPrefix("TableWorkerRegistar", ReplicationId, TargetId)
+        , LogPrefix("TableWorkerStopper", ReplicationId, TargetId)
     {
     }
 
@@ -185,7 +185,7 @@ class TTableWorkerStoper: public TActorBootstrapped<TTableWorkerStoper> {
     const TActorLogPrefix LogPrefix;
     THashSet<TWorkerId> WorkersToStop;
 
-}; // TTableWorkerStoper
+}; // TTableWorkerStopper
 
 TTableTarget::TTableTarget(TReplication* replication, ui64 id, const TString& srcPath, const TString& dstPath)
     : TTargetWithStream(replication, ETargetKind::Table, id, srcPath, dstPath)
@@ -199,9 +199,9 @@ IActor* TTableTarget::CreateWorkerRegistar(const TActorContext& ctx) const {
         CanonizePath(ChildPath(SplitPath(GetSrcPath()), GetStreamName())), GetDstPathId());
 }
 
-IActor* TTableTarget::CreateWorkerStoper(const TActorContext& ctx) const {
+IActor* TTableTarget::CreateWorkerStopper(const TActorContext& ctx) const {
     auto replication = GetReplication();
-    return new TTableWorkerStoper(ctx.SelfID, replication->GetYdbProxy(),
+    return new TTableWorkerStopper(ctx.SelfID, replication->GetYdbProxy(),
         replication->GetConfig().GetSrcConnectionParams(), replication->GetId(), GetId(),
         CanonizePath(ChildPath(SplitPath(GetSrcPath()), GetStreamName())), GetDstPathId());
 }

ydb/core/tx/replication/controller/target_table.h

@@ -11,7 +11,7 @@ class TTableTarget: public TTargetWithStream {
 
 protected:
     IActor* CreateWorkerRegistar(const TActorContext& ctx) const override;
-    IActor* CreateWorkerStoper(const TActorContext& ctx) const override;
+    IActor* CreateWorkerStopper(const TActorContext& ctx) const override;
 
 }; // TTableTarget