Merge 7962922 into 9480de8

Author: FloatingCrowbar

ydb/core/persqueue/pq_impl.cpp

@@ -4621,7 +4621,6 @@ TActorId TPersQueue::GetPartitionQuoter(const TPartitionId& partition) {
             partition,
             SelfId(),
             TabletID(),
-            IsLocalDC,
             *Counters
         ));
     }

ydb/core/persqueue/read_quoter.cpp

@@ -35,7 +35,6 @@ void TPartitionQuoterBase::StartQuoting(TRequestContext&& context) {
     auto& request = context.Request;
     auto* accountQuotaTracker = GetAccountQuotaTracker(request);
     if (accountQuotaTracker) {
-
         Send(accountQuotaTracker->Actor, new NAccountQuoterEvents::TEvRequest(request->Cookie, request->Request.Release()));
         PendingAccountQuotaRequests[request->Cookie] = std::move(context);
     } else {
@@ -115,6 +114,7 @@ void TPartitionQuoterBase::ProcessInflightQueue() {
 
 void TPartitionQuoterBase::HandleConfigUpdate(TEvPQ::TEvChangePartitionConfig::TPtr& ev, const TActorContext& ctx) {
     PQTabletConfig = ev->Get()->Config;
+    TopicConverter = ev->Get()->TopicConverter;
     bool totalQuotaUpdated = false;
     if (PartitionTotalQuotaTracker.Defined()) {
         totalQuotaUpdated = PartitionTotalQuotaTracker->UpdateConfigIfChanged(
@@ -150,6 +150,8 @@ void TReadQuoter::OnAccountQuotaApproved(TRequestContext&& context) {
 }
 
 TAccountQuoterHolder* TReadQuoter::GetAccountQuotaTracker(const THolder<TEvPQ::TEvRequestQuota>& request) {
+    if (!TopicConverter)
+        return nullptr;
     auto clientId = request->Request->CastAsLocal<TEvPQ::TEvRead>()->ClientId;
     return GetOrCreateConsumerQuota(clientId, ActorContext())->AccountQuotaTracker.Get();
 }
@@ -283,6 +285,7 @@ ui64 TReadQuoter::GetTotalPartitionSpeedBurst(const NKikimrPQ::TPQTabletConfig&
 THolder<TAccountQuoterHolder> TReadQuoter::CreateAccountQuotaTracker(const TString& user, const TActorContext& ctx) const {
     const auto& quotingConfig = AppData()->PQConfig.GetQuotingConfig();
     TActorId actorId;
+    Y_ENSURE(TopicConverter);
     if (GetTabletActor() && quotingConfig.GetEnableQuoting()) {
         if(quotingConfig.GetEnableReadQuoting()) {
             actorId = TActivationContext::Register(

ydb/core/persqueue/read_quoter.h

@@ -297,7 +297,6 @@ class TWriteQuoter : public TPartitionQuoterBase {
         const TPartitionId& partition,
         TActorId tabletActor,
         ui64 tabletId,
-        bool isLocalDc,
         const TTabletCountersBase& counters
     );
 
@@ -332,8 +331,8 @@ class TWriteQuoter : public TPartitionQuoterBase {
     }
 
 private:
+    bool GetAccountQuotingEnabled(const NKikimrPQ::TPQConfig& pqConfig) const;
     bool QuotingEnabled;
-    bool AccountQuotingEnabled;
     THolder<TAccountQuoterHolder> AccountQuotaTracker;
 };
 

ydb/core/persqueue/ut/partition_ut.cpp

@@ -368,7 +368,6 @@ TPartition* TPartitionFixture::CreatePartitionActor(const TPartitionId& id,
                 id,
                 Ctx->Edge,
                 Ctx->TabletId,
-                Config.GetLocalDC(),
                 *TabletCounters
         ));
     }

ydb/core/persqueue/write_quoter.cpp

@@ -10,15 +10,13 @@ TWriteQuoter::TWriteQuoter(
     const TPartitionId& partition,
     TActorId tabletActor,
     ui64 tabletId,
-    bool isLocalDc,
     const TTabletCountersBase& counters
 )
     : TPartitionQuoterBase(
             topicConverter, config, partition, tabletActor, pqConfig.GetQuotingConfig().GetEnableQuoting(),
             tabletId, counters, 1
     )
     , QuotingEnabled(pqConfig.GetQuotingConfig().GetEnableQuoting())
-    , AccountQuotingEnabled(IsQuotingEnabled(pqConfig, isLocalDc))
 {
 }
 
@@ -44,6 +42,10 @@ void TWriteQuoter::HandleConsumedImpl(TEvPQ::TEvConsumed::TPtr& ev) {
     }
 }
 
+bool TWriteQuoter::GetAccountQuotingEnabled(const NKikimrPQ::TPQConfig& pqConfig) const {
+    return IsQuotingEnabled(pqConfig, PQTabletConfig.GetLocalDC());
+}
+
 void TWriteQuoter::HandleWakeUpImpl() {
 }
 
@@ -53,7 +55,7 @@ void TWriteQuoter::UpdateQuotaConfigImpl(bool, const TActorContext&) {
 
 THolder<TAccountQuoterHolder> TWriteQuoter::CreateAccountQuotaTracker() const {
     TActorId actorId;
-    if (GetTabletActor() && AccountQuotingEnabled) {
+    if (GetTabletActor() && GetAccountQuotingEnabled(AppData()->PQConfig)) {
         actorId = TActivationContext::Register(
             new TAccountWriteQuoter(
                 GetTabletActor(),
@@ -97,9 +99,12 @@ IEventBase* TWriteQuoter::MakeQuotaApprovedEvent(TRequestContext& context) {
 };
 
 TAccountQuoterHolder* TWriteQuoter::GetAccountQuotaTracker(const THolder<TEvPQ::TEvRequestQuota>&) {
-    if (!AccountQuotingEnabled && !QuotingEnabled) {
+    if (!GetAccountQuotingEnabled(AppData()->PQConfig) && !QuotingEnabled) {
         return nullptr;
     }
+    if (!TopicConverter)
+        return nullptr;
+
     if (!AccountQuotaTracker) {
         AccountQuotaTracker = CreateAccountQuotaTracker();
     }