Merge 4787e02 into 590c21c

niksaveliev · web-flow · commit e3a492a49fa5 · 2024-02-12T13:17:12.000+06:00
diff --git a/ydb/core/persqueue/partition.cpp b/ydb/core/persqueue/partition.cpp
@@ -748,7 +748,7 @@ void TPartition::Handle(TEvPQ::TEvPartitionStatus::TPtr& ev, const TActorContext
             auto& userInfo = userInfoPair.second;
             if (!userInfo.LabeledCounters)
                 continue;
-            if (!userInfo.HasReadRule && !userInfo.Important)
+            if (userInfoPair.first != CLIENTID_WITHOUT_CONSUMER && !userInfo.HasReadRule && !userInfo.Important)
                 continue;
             auto* cac = ac->AddConsumerAggregatedCounters();
             cac->SetConsumer(userInfo.User);
@@ -1083,7 +1083,7 @@ bool TPartition::UpdateCounters(const TActorContext& ctx, bool force) {
         auto& userInfo = userInfoPair.second;
         if (!userInfo.LabeledCounters)
             continue;
-        if (!userInfo.HasReadRule && !userInfo.Important)
+        if (userInfoPair.first != CLIENTID_WITHOUT_CONSUMER && !userInfo.HasReadRule && !userInfo.Important)
             continue;
         bool haveChanges = false;
         userInfo.EndOffset = EndOffset;
@@ -1805,7 +1805,7 @@ void TPartition::OnProcessTxsAndUserActsWriteComplete(ui64 cookie, const TActorC
             } else {
                 TabletCounters.Cumulative()[COUNTER_PQ_WRITE_TIMESTAMP_CACHE_HIT].Increment(1);
             }
-        } else {
+        } else if (user != CLIENTID_WITHOUT_CONSUMER) {
             auto ui = UsersInfoStorage->GetIfExists(user);
             if (ui && ui->LabeledCounters) {
                 ScheduleDropPartitionLabeledCounters(ui->LabeledCounters->GetGroup());
diff --git a/ydb/core/persqueue/partition_init.cpp b/ydb/core/persqueue/partition_init.cpp
@@ -886,7 +886,7 @@ void TPartition::SetupStreamCounters(const TActorContext& ctx) {
     SLIBigLatency = NKikimr::NPQ::TMultiCounter(subGroup, aggr, {}, {"WriteBigLatency"}, true, "name", false);
     WritesTotal = NKikimr::NPQ::TMultiCounter(subGroup, aggr, {}, {"WritesTotal"}, true, "name", false);
     if (IsQuotingEnabled() && !TopicWriteQuotaResourcePath.empty()) {
-        subgroups.push_back({"name", "api.grpc.topic.stream_write.topic_throttled_milliseconds"});
+        subgroups.push_back({"name", "topic.write.topic_throttled_milliseconds"});
         TopicWriteQuotaWaitCounter = THolder<NKikimr::NPQ::TPercentileCounter>(
             new NKikimr::NPQ::TPercentileCounter(
                 NPersQueue::GetCountersForTopic(counters, IsServerless), {},
@@ -899,7 +899,7 @@ void TPartition::SetupStreamCounters(const TActorContext& ctx) {
         subgroups.pop_back();
     }
 
-    subgroups.push_back({"name", "api.grpc.topic.stream_write.partition_throttled_milliseconds"});
+    subgroups.push_back({"name", "topic.write.partition_throttled_milliseconds"});
     PartitionWriteQuotaWaitCounter = THolder<NKikimr::NPQ::TPercentileCounter>(
         new NKikimr::NPQ::TPercentileCounter(
             NPersQueue::GetCountersForTopic(counters, IsServerless), {}, subgroups, "bin",
diff --git a/ydb/core/persqueue/read_balancer.cpp b/ydb/core/persqueue/read_balancer.cpp
@@ -1,6 +1,7 @@
 #include "read_balancer.h"
 
 #include <ydb/core/persqueue/events/internal.h>
+#include "ydb/core/persqueue/user_info.h"
 #include <ydb/core/protos/counters_pq.pb.h>
 #include <ydb/core/base/feature_flags.h>
 #include <ydb/core/tablet/tablet_exception.h>
@@ -65,6 +66,7 @@ bool TPersQueueReadBalancer::TTxInit::Execute(TTransactionContext& txc, const TA
                 for (const auto& rr : Self->TabletConfig.GetReadRules()) {
                     Self->Consumers[rr];
                 }
+                Self->Consumers[CLIENTID_WITHOUT_CONSUMER];
             }
             Self->Inited = true;
             if (!dataRowset.Next())
@@ -536,7 +538,8 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvUpdateBalancerConfig::TPtr
             Consumers[rr];
         }
     }
-
+    Consumers[CLIENTID_WITHOUT_CONSUMER];
+    
     TVector<std::pair<ui32, TPartInfo>> newPartitions;
     TVector<ui32> deletedPartitions;
     TVector<std::pair<ui64, TTabletInfo>> newTablets;
diff --git a/ydb/public/lib/ydb_cli/commands/topic_operations_scenario.cpp b/ydb/public/lib/ydb_cli/commands/topic_operations_scenario.cpp
@@ -206,6 +206,7 @@ void TTopicOperationsScenario::StartConsumerThreads(std::vector<std::future<void
                 .UseTopicCommit = OnlyTableInTx,
                 .UseTableSelect = UseTableSelect && !OnlyTopicInTx,
                 .UseTableUpsert = !OnlyTopicInTx,
+                .ReadWithoutConsumer = ReadWithoutConsumer,
                 .CommitPeriod = CommitPeriod,
                 .CommitMessages = CommitMessages
             };
diff --git a/ydb/public/lib/ydb_cli/commands/topic_operations_scenario.h b/ydb/public/lib/ydb_cli/commands/topic_operations_scenario.h
@@ -70,6 +70,7 @@ class TTopicOperationsScenario {
     bool OnlyTopicInTx = false;
     bool OnlyTableInTx = false;
     bool UseTableSelect = true;
+    bool ReadWithoutConsumer = false;
 
 protected:
     void CreateTopic(const TString& database,
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_reader.cpp b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_reader.cpp
@@ -24,23 +24,33 @@ void TTopicWorkloadReader::ReaderLoop(TTopicWorkloadReaderParams& params, TInsta
     auto topicClient = std::make_unique<NYdb::NTopic::TTopicClient>(params.Driver);
     std::optional<TTransactionSupport> txSupport;
 
-    auto consumerName = TCommandWorkloadTopicDescribe::GenerateConsumerName(params.ConsumerPrefix, params.ConsumerIdx);
     auto describeTopicResult = TCommandWorkloadTopicDescribe::DescribeTopic(params.Database, params.TopicName, params.Driver);
-    auto consumers = describeTopicResult.GetConsumers();
+    NYdb::NTopic::TReadSessionSettings settings;
+
+    if (!params.ReadWithoutConsumer) {
+        auto consumerName = TCommandWorkloadTopicDescribe::GenerateConsumerName(params.ConsumerPrefix, params.ConsumerIdx);
+        auto consumers = describeTopicResult.GetConsumers();
 
-    if (!std::any_of(consumers.begin(), consumers.end(), [consumerName](const auto& consumer) { return consumer.GetConsumerName() == consumerName; }))
-    {
-        WRITE_LOG(params.Log, ELogPriority::TLOG_EMERG, TStringBuilder() << "Topic '" << params.TopicName << "' doesn't have a consumer '" << consumerName << "'. Run command 'workload init' with parameter '--consumers'.");
-        exit(EXIT_FAILURE);
+        if (!std::any_of(consumers.begin(), consumers.end(), [consumerName](const auto& consumer) { return consumer.GetConsumerName() == consumerName; }))
+        {
+            WRITE_LOG(params.Log, ELogPriority::TLOG_EMERG, TStringBuilder() << "Topic '" << params.TopicName << "' doesn't have a consumer '" << consumerName << "'. Run command 'workload init' with parameter '--consumers'.");
+            exit(EXIT_FAILURE);
+        }
+        settings.ConsumerName(consumerName).AppendTopics(params.TopicName);
+    } else {
+        NYdb::NTopic::TTopicReadSettings topic = params.TopicName;
+        auto partitions = describeTopicResult.GetPartitions();
+        for(auto partition: partitions) {
+            topic.AppendPartitionIds(partition.GetPartitionId());
+        }
+        settings.WithoutConsumer().AppendTopics(topic);
     }
     
+    
     if (params.UseTransactions) {
         txSupport.emplace(params.Driver, params.ReadOnlyTableName, params.TableName);
     }
 
-    NYdb::NTopic::TReadSessionSettings settings;
-    settings.ConsumerName(consumerName).AppendTopics(params.TopicName);
-
     auto readSession = topicClient->CreateReadSession(settings);
     WRITE_LOG(params.Log, ELogPriority::TLOG_INFO, "Reader session was created.");
 
@@ -93,7 +103,7 @@ void TTopicWorkloadReader::ReaderLoop(TTopicWorkloadReaderParams& params, TInsta
                         << " createTime " << message.GetCreateTime() << " fullTimeMs " << fullTime);
                 }
 
-                if (!txSupport || params.UseTopicCommit) {
+                if (!params.ReadWithoutConsumer && (!txSupport || params.UseTopicCommit)) {
                     dataEvent->Commit();
                 }
             } else if (auto* createPartitionStreamEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TStartPartitionSessionEvent>(&event)) {
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_reader.h b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_reader.h
@@ -30,6 +30,7 @@ namespace NYdb {
             bool UseTopicCommit = false;
             bool UseTableSelect = true;
             bool UseTableUpsert = true;
+            bool ReadWithoutConsumer = false;
             size_t CommitPeriod = 15;
             size_t CommitMessages = 1'000'000;
         };
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.cpp b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.cpp
@@ -34,6 +34,9 @@ void TCommandWorkloadTopicRunRead::Config(TConfig& config)
     config.Opts->AddLongOption("topic", "Topic name.")
         .DefaultValue(TOPIC)
         .StoreResult(&Scenario.TopicName);
+    config.Opts->AddLongOption("no-consumer", "Read without consumer")
+        .Hidden()
+        .StoreTrue(&Scenario.ReadWithoutConsumer);
 
     // Specific params
     config.Opts->AddLongOption("consumer-prefix", "Use consumers with names '<consumer-prefix>-0' ... '<consumer-prefix>-<n-1>' where n is set in the '--consumers' option.")

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`#include "read_balancer.h"`
`2`	`2`
`3`	`3`	`#include <ydb/core/persqueue/events/internal.h>`
	`4`	`+#include "ydb/core/persqueue/user_info.h"`
`4`	`5`	`#include <ydb/core/protos/counters_pq.pb.h>`
`5`	`6`	`#include <ydb/core/base/feature_flags.h>`
`6`	`7`	`#include <ydb/core/tablet/tablet_exception.h>`
`@@ -65,6 +66,7 @@ bool TPersQueueReadBalancer::TTxInit::Execute(TTransactionContext& txc, const TA`
`65`	`66`	`for (const auto& rr : Self->TabletConfig.GetReadRules()) {`
`66`	`67`	`Self->Consumers[rr];`
`67`	`68`	`}`
	`69`	`+ Self->Consumers[CLIENTID_WITHOUT_CONSUMER];`
`68`	`70`	`}`
`69`	`71`	`Self->Inited = true;`
`70`	`72`	`if (!dataRowset.Next())`
`@@ -536,7 +538,8 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvUpdateBalancerConfig::TPtr`
`536`	`538`	`Consumers[rr];`
`537`	`539`	`}`
`538`	`540`	`}`
`539`		`-`
	`541`	`+ Consumers[CLIENTID_WITHOUT_CONSUMER];`
	`542`	`+`
`540`	`543`	`TVector<std::pair<ui32, TPartInfo>> newPartitions;`
`541`	`544`	`TVector<ui32> deletedPartitions;`
`542`	`545`	`TVector<std::pair<ui64, TTabletInfo>> newTablets;`