Fixed StartProducerWithCDSAndPreferUnknownCluster test (#17585)

Author: nshestakov

ydb/services/deprecated/persqueue_v0/grpc_pq_clusters_updater_actor.cpp

@@ -11,8 +11,9 @@ namespace NGRpcProxy {
 static const int CLUSTERS_UPDATER_TIMEOUT_ON_ERROR = 1;
 
 
-TClustersUpdater::TClustersUpdater(IPQClustersUpdaterCallback* callback)
+TClustersUpdater::TClustersUpdater(IPQClustersUpdaterCallback* callback, TStatus::TPtr& status)
     : Callback(callback)
+    , Status(status)
     {};
 
 void TClustersUpdater::Bootstrap(const NActors::TActorContext& ctx) {
@@ -34,15 +35,21 @@ void TClustersUpdater::Handle(TEvPQClustersUpdater::TEvUpdateClusters::TPtr&, co
     ctx.Send(NKqp::MakeKqpProxyID(ctx.SelfID.NodeId()), req.Release());
 }
 
-void TClustersUpdater::Handle(NNetClassifier::TEvNetClassifier::TEvClassifierUpdate::TPtr& ev, const TActorContext&) {
+void TClustersUpdater::Handle(NNetClassifier::TEvNetClassifier::TEvClassifierUpdate::TPtr& ev, const TActorContext& ctx) {
+    TGuard<TSpinLock> guard(Status->Lock);
+    if (!Status->Running) {
+        return Die(ctx);
+    }
 
     Callback->NetClassifierUpdated(ev->Get()->Classifier);
 }
 
-
-
-
 void TClustersUpdater::Handle(NKqp::TEvKqp::TEvQueryResponse::TPtr &ev, const TActorContext &ctx) {
+    TGuard<TSpinLock> guard(Status->Lock);
+    if (!Status->Running) {
+        return Die(ctx);
+    }
+
     auto& record = ev->Get()->Record;
 
     if (record.GetYdbStatus() == Ydb::StatusIds::SUCCESS) {

ydb/services/deprecated/persqueue_v0/grpc_pq_clusters_updater_actor.h

@@ -47,7 +47,19 @@ class IPQClustersUpdaterCallback {
 
 class TClustersUpdater : public NActors::TActorBootstrapped<TClustersUpdater> {
 public:
-    TClustersUpdater(IPQClustersUpdaterCallback* callback);
+    struct TStatus {
+        using TPtr = std::shared_ptr<TStatus>;
+
+        bool Running = true;
+        TSpinLock Lock;
+
+        void Stop() {
+            TGuard guard(Lock);
+            Running = false;
+        }
+    };
+
+    TClustersUpdater(IPQClustersUpdaterCallback* callback, TStatus::TPtr& status);
 
     void Bootstrap(const NActors::TActorContext& ctx);
 
@@ -58,6 +70,7 @@ class TClustersUpdater : public NActors::TActorBootstrapped<TClustersUpdater> {
     TString LocalCluster;
     TVector<TString> Clusters;
     bool Enabled = false;
+    TStatus::TPtr Status;
 
     STFUNC(StateFunc) {
         switch (ev->GetTypeRewrite()) {

ydb/services/deprecated/persqueue_v0/grpc_pq_read.cpp

@@ -193,7 +193,8 @@ TPQReadService::TPQReadService(NKikimr::NGRpcService::TGRpcPersQueueService* ser
     );
 
     if (NeedDiscoverClusters) {
-        ActorSystem->Register(new TClustersUpdater(this));
+        ClustersUpdaterStatus = std::make_shared<TClustersUpdater::TStatus>();
+        ActorSystem->Register(new TClustersUpdater(this, ClustersUpdaterStatus));
     }
 }
 

ydb/services/deprecated/persqueue_v0/grpc_pq_read.h

@@ -79,6 +79,9 @@ class TPQReadService : public IPQClustersUpdaterCallback, public std::enable_sha
 
     void StopService() {
         AtomicSet(ShuttingDown_, 1);
+        if (ClustersUpdaterStatus) {
+            ClustersUpdaterStatus->Stop();
+        }
     }
 
     bool IsShuttingDown() const {
@@ -139,6 +142,8 @@ class TPQReadService : public IPQClustersUpdaterCallback, public std::enable_sha
     NAddressClassifier::TLabeledAddressClassifier::TConstPtr DatacenterClassifier; // Detects client's datacenter by IP. May be null
 
     bool NeedDiscoverClusters;
+    TClustersUpdater::TStatus::TPtr ClustersUpdaterStatus;
+
     NPersQueue::TConverterFactoryPtr TopicConverterFactory;
     std::unique_ptr<NPersQueue::TTopicsListController> TopicsHandler;
 };

ydb/services/deprecated/persqueue_v0/grpc_pq_write.cpp

@@ -151,7 +151,8 @@ void TPQWriteServiceImpl::InitClustersUpdater()
     TAppData* appData = ActorSystem->AppData<TAppData>();
     NeedDiscoverClusters = !appData->PQConfig.GetTopicsAreFirstClassCitizen();
     if (NeedDiscoverClusters) {
-        ActorSystem->Register(new TClustersUpdater(this));
+        ClustersUpdaterStatus = std::make_shared<TClustersUpdater::TStatus>();
+        ActorSystem->Register(new TClustersUpdater(this, ClustersUpdaterStatus));
     }
 }
 

ydb/services/deprecated/persqueue_v0/grpc_pq_write.h

@@ -69,6 +69,9 @@ class TPQWriteServiceImpl : public IPQClustersUpdaterCallback, public std::enabl
 
     void StopService() {
         AtomicSet(ShuttingDown_, 1);
+        if (ClustersUpdaterStatus) {
+            ClustersUpdaterStatus->Stop();
+        }
     }
 
     bool IsShuttingDown() const {
@@ -116,6 +119,7 @@ class TPQWriteServiceImpl : public IPQClustersUpdaterCallback, public std::enabl
     TAtomic ShuttingDown_ = 0;
 
     bool NeedDiscoverClusters; // Legacy mode OR account-mode in multi-cluster setup;
+    TClustersUpdater::TStatus::TPtr ClustersUpdaterStatus;
 
     NAddressClassifier::TLabeledAddressClassifier::TConstPtr DatacenterClassifier; // Detects client's datacenter by IP. May be null
 };