test TPersQueueTest::DirectReadStop (#4546)

Author: Alek5andr-Kotov

.github/config/muted_ya.txt

@@ -62,7 +62,6 @@ ydb/services/persqueue_v1/ut TPersQueueTest.SetupLockSession
 ydb/services/persqueue_v1/ut TPQCompatTest.BadTopics
 ydb/services/persqueue_v1/ut TPersQueueTest.DirectRead*Cache*
 ydb/services/persqueue_v1/ut [3/10]*
-ydb/services/persqueue_v1/ut TPersQueueTest.DirectReadStop
 ydb/services/ydb/sdk_sessions_pool_ut YdbSdkSessionsPool.StressTestSync*
 ydb/services/ydb/sdk_sessions_ut YdbSdkSessions.TestActiveSessionCountAfterBadSession
 ydb/services/ydb/sdk_sessions_ut [6/10]*

ydb/services/persqueue_v1/persqueue_new_schemecache_ut.cpp

@@ -506,7 +506,7 @@ namespace NKikimr::NPersQueueTests {
 
     Y_UNIT_TEST_SUITE(TPersqueueDataPlaneTestSuite) {
         Y_UNIT_TEST(WriteSession) {
-            TPersQueueV1TestServer server(true, true);
+            TPersQueueV1TestServer server({.CheckACL=true, .TenantModeEnabled=true});
 
             TString topic = "/Root/account1/write_topic";
             TString consumer = "consumer_aba";
@@ -570,7 +570,7 @@ namespace NKikimr::NPersQueueTests {
 
     Y_UNIT_TEST_SUITE(TPersqueueControlPlaneTestSuite) {
         Y_UNIT_TEST(SetupReadLockSessionWithDatabase) {
-            TPersQueueV1TestServer server(false, true);
+            TPersQueueV1TestServer server({.TenantModeEnabled=true});
 
             {
                 auto res = server.PersQueueClient->AddReadRule("/Root/acc/topic1", TAddReadRuleSettings().ReadRule(TReadRuleSettings().ConsumerName("user1")));
@@ -606,7 +606,7 @@ namespace NKikimr::NPersQueueTests {
         }
 
         Y_UNIT_TEST(SetupWriteLockSessionWithDatabase) {
-            TPersQueueV1TestServer server(false, true);
+            TPersQueueV1TestServer server({.TenantModeEnabled=true});
 
             auto stub = Ydb::PersQueue::V1::PersQueueService::NewStub(server.InsecureChannel);
             grpc::ClientContext grpcContext;
@@ -631,7 +631,7 @@ namespace NKikimr::NPersQueueTests {
         }
 
         Y_UNIT_TEST(TestAddRemoveReadRule) {
-            TPersQueueV1TestServer server(false, true);
+            TPersQueueV1TestServer server({.TenantModeEnabled=true});
             SET_LOCALS;
 
             pqClient->CreateConsumer("goodUser");

ydb/services/persqueue_v1/persqueue_ut.cpp

@@ -953,7 +953,7 @@ Y_UNIT_TEST_SUITE(TPersQueueTest) {
     }
 
     Y_UNIT_TEST(DirectReadPreCached) {
-        TPersQueueV1TestServer server{true};
+        TPersQueueV1TestServer server{{.CheckACL=true}};
         SET_LOCALS;
         TDirectReadTestSetup setup{server};
         setup.DoWrite(pqClient->GetDriver(), "acc/topic1", 1_MB, 30);
@@ -981,7 +981,7 @@ Y_UNIT_TEST_SUITE(TPersQueueTest) {
     }
 
     Y_UNIT_TEST(DirectReadNotCached) {
-        TPersQueueV1TestServer server{true};
+        TPersQueueV1TestServer server{{.CheckACL=true}};
         SET_LOCALS;
         TDirectReadTestSetup setup{server};
 
@@ -1013,7 +1013,7 @@ Y_UNIT_TEST_SUITE(TPersQueueTest) {
     }
 
     Y_UNIT_TEST(DirectReadBadCases) {
-        TPersQueueV1TestServer server{true};
+        TPersQueueV1TestServer server{{.CheckACL=true}};
         SET_LOCALS;
         TDirectReadTestSetup setup{server};
         setup.InitControlSession("acc/topic1");
@@ -1042,7 +1042,7 @@ Y_UNIT_TEST_SUITE(TPersQueueTest) {
     }
 
     Y_UNIT_TEST(DirectReadStop) {
-        TPersQueueV1TestServer server{true};
+        TPersQueueV1TestServer server{{.CheckACL=true, .NodeCount=1}};
         SET_LOCALS;
 
         server.Server->AnnoyingClient->AlterTopicNoLegacy("Root/PQ/rt3.dc1--acc--topic1", 2);

ydb/services/persqueue_v1/ut/persqueue_common_tests.h

@@ -51,7 +51,7 @@ class TCommonTests {
     {}
 
     TPersQueueV1TestServer CreateServer() {
-        return TPersQueueV1TestServer(false, TenantModeEnabled);
+        return TPersQueueV1TestServer({.TenantModeEnabled=TenantModeEnabled});
     }
 
     TPersQueueV1TestServerWithRateLimiter CreateServerWithRateLimiter() {

ydb/services/persqueue_v1/ut/persqueue_test_fixture.h

@@ -60,10 +60,13 @@ static void ModifyTopicACL(NYdb::TDriver* driver, const TString& topic, const TV
             Y_UNUSED(settings);
         }
 
+        virtual NKikimr::Tests::TServerSettings GetServerSettings() {
+            return NKikimr::NPersQueueTests::PQSettings();
+        }
 
         void InitializePQ() {
             Y_ABORT_UNLESS(Server == nullptr);
-            Server = MakeHolder<NPersQueue::TTestServer>(false);
+            Server = MakeHolder<NPersQueue::TTestServer>(GetServerSettings(), false);
             Server->ServerSettings.PQConfig.SetTopicsAreFirstClassCitizen(TenantModeEnabled());
             Server->ServerSettings.PQConfig.MutablePQDiscoveryConfig()->SetLBFrontEnabled(true);
             Server->ServerSettings.PQConfig.SetACLRetryTimeoutSec(1);
@@ -214,11 +217,17 @@ static void ModifyTopicACL(NYdb::TDriver* driver, const TString& topic, const TV
         THolder<NYdb::NPersQueue::TPersQueueClient> PersQueueClient;
     };
 
+    struct TPersQueueV1TestServerSettings {
+        bool CheckACL = false;
+        bool TenantModeEnabled = false;
+        ui32 NodeCount = PQ_DEFAULT_NODE_COUNT;
+    };
+
     class TPersQueueV1TestServer : public TPersQueueV1TestServerBase {
     public:
-        TPersQueueV1TestServer(bool checkAcl = false, bool tenantModeEnabled = false)
-            : TPersQueueV1TestServerBase(tenantModeEnabled)
-            , CheckACL(checkAcl)
+        explicit TPersQueueV1TestServer(const TPersQueueV1TestServerSettings& settings = {})
+            : TPersQueueV1TestServerBase(settings.TenantModeEnabled)
+            , Settings(settings)
         {
             InitAll();
         }
@@ -228,11 +237,18 @@ static void ModifyTopicACL(NYdb::TDriver* driver, const TString& topic, const TV
         }
 
         void AlterSettings(NKikimr::Tests::TServerSettings& settings) override {
-            if (CheckACL)
+            if (Settings.CheckACL) {
                 settings.PQConfig.SetCheckACL(true);
+            }
+        }
+
+        NKikimr::Tests::TServerSettings GetServerSettings() override {
+            return NKikimr::NPersQueueTests::PQSettings()
+                .SetNodeCount(Settings.NodeCount);
         }
+
     private:
-        bool CheckACL;
+        TPersQueueV1TestServerSettings Settings;
     };
 
     class TPersQueueV1TestServerWithRateLimiter : public TPersQueueV1TestServerBase {