Skip to content

Commit 3f56c12

Browse files
nshestakovnshestakov
authored
Fixed OutOfSpace tests (#24327)
1 parent 2f955be commit 3f56c12

File tree

5 files changed

+21
-25
lines changed

5 files changed

+21
-25
lines changed

‎ydb/core/persqueue/partition.cpp

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -808,7 +808,7 @@ void TPartition::Handle(TEvPQ::TEvPartitionStatus::TPtr& ev, const TActorContext
808808
result.SetGeneration(TabletGeneration);
809809
result.SetCookie(++PQRBCookie);
810810

811-
if (DiskIsFull || WaitingForSubDomainQuota(ctx)) {
811+
if (DiskIsFull || WaitingForSubDomainQuota()) {
812812
result.SetStatus(NKikimrPQ::TStatusResponse::STATUS_DISK_IS_FULL);
813813
} else if (BlobEncoder.EndOffset - BlobEncoder.StartOffset >= static_cast<ui64>(Config.GetPartitionConfig().GetMaxCountInPartition()) ||
814814
Size() >= static_cast<ui64>(Config.GetPartitionConfig().GetMaxSizeInPartition())) {
@@ -2000,6 +2000,8 @@ void TPartition::Handle(NReadQuoterEvents::TEvQuotaUpdated::TPtr& ev, const TAct
20002000
}
20012001

20022002
void TPartition::Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorContext& ctx) {
2003+
PQ_LOG_D("Received TEvKeyValue::TEvResponse");
2004+
20032005
auto& response = ev->Get()->Record;
20042006

20052007
if (response.HasCookie() && (response.GetCookie() == static_cast<ui64>(ERequestCookie::CompactificationWrite))) {

‎ydb/core/persqueue/partition.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -320,7 +320,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
320320
bool CleanUpBlobs(TEvKeyValue::TEvRequest *request, const TActorContext& ctx);
321321
bool IsQuotingEnabled() const;
322322
bool WaitingForPreviousBlobQuota() const;
323-
bool WaitingForSubDomainQuota(const TActorContext& ctx, const ui64 withSize = 0) const;
323+
bool WaitingForSubDomainQuota(const ui64 withSize = 0) const;
324324
size_t GetQuotaRequestSize(const TEvKeyValue::TEvRequest& request);
325325
std::pair<TInstant, TInstant> GetTime(const TUserInfo& userInfo, ui64 offset) const;
326326
ui32 NextChannel(bool isHead, ui32 blobSize);
@@ -522,7 +522,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
522522
void Bootstrap(const TActorContext& ctx);
523523

524524
ui64 Size() const {
525-
return CompactionBlobEncoder.GetSize();
525+
return CompactionBlobEncoder.GetSize() + BlobEncoder.GetSize();
526526
}
527527

528528
// The size of the data realy was persisted in the storage by the partition

‎ydb/core/persqueue/partition_monitoring.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ void TPartition::HandleMonitoring(TEvPQ::TEvMonRequest::TPtr& ev, const TActorCo
5252
PROPERTIES("Status") {
5353
PROPERTY("State", NKikimrPQ::ETopicPartitionStatus_Name(PartitionConfig->GetStatus()));
5454
PROPERTY("Disk", (DiskIsFull ? "Full" : "Normal"));
55-
PROPERTY("Quota", (WaitingForSubDomainQuota(ctx) ? "Out of space" : "Normal"));
55+
PROPERTY("Quota", (WaitingForSubDomainQuota() ? "Out of space" : "Normal"));
5656
}
5757

5858
PROPERTIES("Information") {

‎ydb/core/persqueue/partition_write.cpp

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -193,7 +193,7 @@ void TPartition::ProcessReserveRequests(const TActorContext& ctx) {
193193
break;
194194
}
195195

196-
if (WaitingForSubDomainQuota(ctx, currentSize)) {
196+
if (WaitingForSubDomainQuota(currentSize)) {
197197
PQ_LOG_D("Reserve processing: SubDomainOutOfSpace. Partition: " << Partition);
198198
break;
199199
}
@@ -486,7 +486,7 @@ void TPartition::OnHandleWriteResponse(const TActorContext& ctx)
486486

487487
void TPartition::Handle(TEvPQ::TEvHandleWriteResponse::TPtr&, const TActorContext& ctx)
488488
{
489-
PQ_LOG_T("TPartition::Handle TEvHandleWriteResponse.");
489+
PQ_LOG_D("Received TPartition::Handle TEvHandleWriteResponse.");
490490
OnHandleWriteResponse(ctx);
491491
}
492492

@@ -657,7 +657,7 @@ void TPartition::ChangeScaleStatusIfNeeded(NKikimrPQ::EScaleStatus scaleStatus)
657657
}
658658

659659
void TPartition::HandleOnWrite(TEvPQ::TEvWrite::TPtr& ev, const TActorContext& ctx) {
660-
PQ_LOG_T("TPartition::TEvWrite");
660+
PQ_LOG_D("Received TPartition::TEvWrite");
661661

662662
if (!CanEnqueue()) {
663663
ReplyError(ctx, ev->Get()->Cookie, InactivePartitionErrorCode,
@@ -771,7 +771,7 @@ void TPartition::HandleOnWrite(TEvPQ::TEvWrite::TPtr& ev, const TActorContext& c
771771
++*offset;
772772
}
773773
}
774-
if (WaitingForPreviousBlobQuota() || WaitingForSubDomainQuota(ctx)) {
774+
if (WaitingForPreviousBlobQuota() || WaitingForSubDomainQuota()) {
775775
SetDeadlinesForWrites(ctx);
776776
}
777777
WriteInflightSize += size;
@@ -1021,19 +1021,20 @@ void TPartition::ExecRequest(TSplitMessageGroupMsg& msg, ProcessParameters& para
10211021

10221022
TPartition::EProcessResult TPartition::PreProcessRequest(TWriteMsg& p) {
10231023
if (!CanWrite()) {
1024-
WriteInflightSize -= p.Msg.Data.size();
1024+
WriteInflightSize -= p.Msg.Data.size();
10251025
ScheduleReplyError(p.Cookie, false, InactivePartitionErrorCode,
10261026
TStringBuilder() << "Write to inactive partition " << Partition.OriginalPartitionId);
10271027
return EProcessResult::ContinueDrop;
10281028
}
10291029

10301030
if (DiskIsFull) {
1031-
WriteInflightSize -= p.Msg.Data.size();
1031+
WriteInflightSize -= p.Msg.Data.size();
10321032
ScheduleReplyError(p.Cookie, false,
10331033
NPersQueue::NErrorCode::WRITE_ERROR_DISK_IS_FULL,
10341034
"Disk is full");
10351035
return EProcessResult::ContinueDrop;
10361036
}
1037+
10371038
if (TxAffectedSourcesIds.contains(p.Msg.SourceId)) {
10381039
return EProcessResult::Blocked;
10391040
}
@@ -1639,7 +1640,7 @@ bool TPartition::RequestBlobQuota()
16391640

16401641
void TPartition::HandlePendingRequests(const TActorContext& ctx)
16411642
{
1642-
if (WaitingForPreviousBlobQuota() || WaitingForSubDomainQuota(ctx) || NeedDeletePartition) {
1643+
if (WaitingForPreviousBlobQuota() || WaitingForSubDomainQuota() || NeedDeletePartition) {
16431644
return;
16441645
}
16451646
if (RequestBlobQuota()) {
@@ -1810,7 +1811,7 @@ bool TPartition::WaitingForPreviousBlobQuota() const {
18101811
return TopicQuotaRequestCookie != 0;
18111812
}
18121813

1813-
bool TPartition::WaitingForSubDomainQuota(const TActorContext& /*ctx*/, const ui64 withSize) const {
1814+
bool TPartition::WaitingForSubDomainQuota(const ui64 withSize) const {
18141815
if (!SubDomainOutOfSpace || !AppData()->FeatureFlags.GetEnableTopicDiskSubDomainQuota()) {
18151816
return false;
18161817
}

‎ydb/core/persqueue/ut/partition_ut.cpp

Lines changed: 6 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1266,6 +1266,8 @@ void TPartitionFixture::TestWriteSubDomainOutOfSpace_DeadlineWork(bool ignoreQuo
12661266
{
12671267
Ctx->Runtime->GetAppData().FeatureFlags.SetEnableTopicDiskSubDomainQuota(true);
12681268
Ctx->Runtime->GetAppData().PQConfig.MutableQuotingConfig()->SetQuotaWaitDurationMs(300);
1269+
Ctx->Runtime->SetLogPriority( NKikimrServices::PERSQUEUE, NActors::NLog::PRI_DEBUG);
1270+
12691271
CreatePartition({
12701272
.Partition=TPartitionId{1},
12711273
.Begin=0, .End=0,
@@ -1293,10 +1295,9 @@ void TPartitionFixture::TestWriteSubDomainOutOfSpace_DeadlineWork(bool ignoreQuo
12931295
return cookie == e.Cookie;
12941296
};
12951297

1296-
TString data = "data for write";
1297-
12981298
// First message will be processed because used storage 0 and limit 0. That is, the limit is not exceeded.
1299-
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data, ignoreQuotaDeadline);
1299+
TString data0 = "data for write 0";
1300+
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data0, ignoreQuotaDeadline);
13001301
messageNo++;
13011302

13021303
WaitKeyValueRequest(kvCookie); // the partition saves the TEvPQ::TEvWrite event
@@ -1308,7 +1309,8 @@ void TPartitionFixture::TestWriteSubDomainOutOfSpace_DeadlineWork(bool ignoreQuo
13081309
}
13091310

13101311
// Second message will not be processed because the limit is exceeded.
1311-
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data, ignoreQuotaDeadline);
1312+
TString data1 = "data for write 1";
1313+
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data1, ignoreQuotaDeadline);
13121314
messageNo++;
13131315

13141316
{
@@ -2559,25 +2561,16 @@ Y_UNIT_TEST_F(ReserveSubDomainOutOfSpace, TPartitionFixture)
25592561

25602562
Y_UNIT_TEST_F(WriteSubDomainOutOfSpace, TPartitionFixture)
25612563
{
2562-
// TODO(abcdef): temporarily deleted
2563-
return;
2564-
25652564
TestWriteSubDomainOutOfSpace_DeadlineWork(false);
25662565
}
25672566

25682567
Y_UNIT_TEST_F(WriteSubDomainOutOfSpace_DisableExpiration, TPartitionFixture)
25692568
{
2570-
// TODO(abcdef): temporarily deleted
2571-
return;
2572-
25732569
TestWriteSubDomainOutOfSpace(TDuration::MilliSeconds(0), false);
25742570
}
25752571

25762572
Y_UNIT_TEST_F(WriteSubDomainOutOfSpace_IgnoreQuotaDeadline, TPartitionFixture)
25772573
{
2578-
// TODO(abcdef): temporarily deleted
2579-
return;
2580-
25812574
TestWriteSubDomainOutOfSpace_DeadlineWork(true);
25822575
}
25832576

0 commit comments

Comments
 (0)