Skip to content

Reply error on database size exceeded #12110

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Nov 29, 2024
Merged

Reply error on database size exceeded #12110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dd6d2f3
Reply error on database size exceeded
nshestakov Nov 28, 2024
435465f
fix test
nshestakov Nov 28, 2024
aa089a3
fix
nshestakov Nov 28, 2024
0ae7a06
fix
nshestakov Nov 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions ydb/core/persqueue/partition_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ namespace NKikimr::NPQ {

static const ui32 BATCH_UNPACK_SIZE_BORDER = 500_KB;
static const ui32 MAX_INLINE_SIZE = 1000;
static const TDuration SubDomainQuotaWaitDurationMs = TDuration::Seconds(60);

static constexpr NPersQueue::NErrorCode::EErrorCode InactivePartitionErrorCode = NPersQueue::NErrorCode::WRITE_ERROR_PARTITION_INACTIVE;

Expand Down Expand Up @@ -1483,8 +1484,12 @@ void TPartition::AddNewWriteBlob(std::pair<TKey, ui32>& res, TEvKeyValue::TEvReq

void TPartition::SetDeadlinesForWrites(const TActorContext& ctx) {
PQ_LOG_T("TPartition::SetDeadlinesForWrites.");
if (AppData(ctx)->PQConfig.GetQuotingConfig().GetQuotaWaitDurationMs() > 0 && QuotaDeadline == TInstant::Zero()) {
QuotaDeadline = ctx.Now() + TDuration::MilliSeconds(AppData(ctx)->PQConfig.GetQuotingConfig().GetQuotaWaitDurationMs());
auto quotaWaitDurationMs = TDuration::MilliSeconds(AppData(ctx)->PQConfig.GetQuotingConfig().GetQuotaWaitDurationMs());
if (SubDomainOutOfSpace) {
quotaWaitDurationMs = quotaWaitDurationMs ? std::min(quotaWaitDurationMs, SubDomainQuotaWaitDurationMs) : SubDomainQuotaWaitDurationMs;
}
if (quotaWaitDurationMs > TDuration::Zero() && QuotaDeadline == TInstant::Zero()) {
QuotaDeadline = ctx.Now() + quotaWaitDurationMs;

ctx.Schedule(QuotaDeadline, new TEvPQ::TEvQuotaDeadlineCheck());
}
Expand Down Expand Up @@ -1513,7 +1518,7 @@ void TPartition::FilterDeadlinedWrites(const TActorContext& ctx, TMessageQueue&
{
TMessageQueue newRequests;
for (auto& w : requests) {
if (!w.IsWrite() || w.GetWrite().Msg.IgnoreQuotaDeadline) {
if (!w.IsWrite() || (w.GetWrite().Msg.IgnoreQuotaDeadline && !SubDomainOutOfSpace)) {
newRequests.emplace_back(std::move(w));
continue;
}
Expand All @@ -1529,7 +1534,8 @@ void TPartition::FilterDeadlinedWrites(const TActorContext& ctx, TMessageQueue&
WriteInflightSize -= msg.Data.size();
}

ReplyError(ctx, w.GetCookie(), NPersQueue::NErrorCode::OVERLOAD, "quota exceeded");
TString errorMsg = SubDomainOutOfSpace ? "database size exceeded" : "quota exceeded";
ReplyError(ctx, w.GetCookie(), NPersQueue::NErrorCode::OVERLOAD, errorMsg);
}
requests = std::move(newRequests);
}
Expand Down
112 changes: 59 additions & 53 deletions ydb/core/persqueue/ut/partition_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,6 +287,7 @@ class TPartitionFixture : public NUnitTest::TBaseFixture {
void ShadowPartitionCountersTest(bool isFirstClass);

void TestWriteSubDomainOutOfSpace(TDuration quotaWaitDuration, bool ignoreQuotaDeadline);
void TestWriteSubDomainOutOfSpace_DeadlineWork(bool ignoreQuotaDeadline);
void WaitKeyValueRequest(TMaybe<ui64>& cookie);

void CmdChangeOwner(ui64 cookie, const TString& sourceId, TDuration duration, TString& ownerCookie);
Expand Down Expand Up @@ -1206,6 +1207,62 @@ void TPartitionFixture::EmulateKVTablet()
Cerr << "Send disk status response with cookie: " << cookie.GetOrElse(0) << Endl;
}

void TPartitionFixture::TestWriteSubDomainOutOfSpace_DeadlineWork(bool ignoreQuotaDeadline)
{
Ctx->Runtime->GetAppData().FeatureFlags.SetEnableTopicDiskSubDomainQuota(true);
Ctx->Runtime->GetAppData().PQConfig.MutableQuotingConfig()->SetQuotaWaitDurationMs(300);
CreatePartition({
.Partition=TPartitionId{1},
.Begin=0, .End=0,
//
// partition configuration
//
.Config={.Version=1, .Consumers={{.Consumer="client-1", .Offset=3}}}
},
//
// tablet configuration
//
{.Version=2, .Consumers={{.Consumer="client-1"}}});
TMaybe<ui64> kvCookie;

SendSubDomainStatus(true);

ui64 cookie = 1;
ui64 messageNo = 0;
TString ownerCookie;

CmdChangeOwner(cookie, "owner1", TDuration::Seconds(1), ownerCookie);

TAutoPtr<IEventHandle> handle;
std::function<bool(const TEvPQ::TEvError&)> truth = [&](const TEvPQ::TEvError& e) {
return cookie == e.Cookie;
};

TString data = "data for write";

// First message will be processed because used storage 0 and limit 0. That is, the limit is not exceeded.
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data, ignoreQuotaDeadline);
messageNo++;

WaitKeyValueRequest(kvCookie); // the partition saves the TEvPQ::TEvWrite event
SendDiskStatusResponse(&kvCookie);

{
auto event = Ctx->Runtime->GrabEdgeEvent<TEvPQ::TEvProxyResponse>(TDuration::Seconds(1));
UNIT_ASSERT(event != nullptr);
}

// Second message will not be processed because the limit is exceeded.
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data, ignoreQuotaDeadline);
messageNo++;

{
auto event = Ctx->Runtime->GrabEdgeEventIf<TEvPQ::TEvError>(handle, truth, TDuration::Seconds(1));
UNIT_ASSERT(event != nullptr);
UNIT_ASSERT_EQUAL(NPersQueue::NErrorCode::OVERLOAD, event->ErrorCode);
}
}

void TPartitionFixture::TestWriteSubDomainOutOfSpace(TDuration quotaWaitDuration, bool ignoreQuotaDeadline)
{
Ctx->Runtime->GetAppData().FeatureFlags.SetEnableTopicDiskSubDomainQuota(true);
Expand Down Expand Up @@ -2323,58 +2380,7 @@ Y_UNIT_TEST_F(ReserveSubDomainOutOfSpace, TPartitionFixture)

Y_UNIT_TEST_F(WriteSubDomainOutOfSpace, TPartitionFixture)
{
Ctx->Runtime->GetAppData().FeatureFlags.SetEnableTopicDiskSubDomainQuota(true);
Ctx->Runtime->GetAppData().PQConfig.MutableQuotingConfig()->SetQuotaWaitDurationMs(300);
CreatePartition({
.Partition=TPartitionId{1},
.Begin=0, .End=0,
//
// partition configuration
//
.Config={.Version=1, .Consumers={{.Consumer="client-1", .Offset=3}}}
},
//
// tablet configuration
//
{.Version=2, .Consumers={{.Consumer="client-1"}}});
TMaybe<ui64> kvCookie;

SendSubDomainStatus(true);

ui64 cookie = 1;
ui64 messageNo = 0;
TString ownerCookie;

CmdChangeOwner(cookie, "owner1", TDuration::Seconds(1), ownerCookie);

TAutoPtr<IEventHandle> handle;
std::function<bool(const TEvPQ::TEvError&)> truth = [&](const TEvPQ::TEvError& e) {
return cookie == e.Cookie;
};

TString data = "data for write";

// First message will be processed because used storage 0 and limit 0. That is, the limit is not exceeded.
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data);
messageNo++;

WaitKeyValueRequest(kvCookie); // the partition saves the TEvPQ::TEvWrite event
SendDiskStatusResponse(&kvCookie);

{
auto event = Ctx->Runtime->GrabEdgeEvent<TEvPQ::TEvProxyResponse>(TDuration::Seconds(1));
UNIT_ASSERT(event != nullptr);
}

// Second message will not be processed because the limit is exceeded.
SendWrite(++cookie, messageNo, ownerCookie, (messageNo + 1) * 100, data);
messageNo++;

{
auto event = Ctx->Runtime->GrabEdgeEventIf<TEvPQ::TEvError>(handle, truth, TDuration::Seconds(1));
UNIT_ASSERT(event != nullptr);
UNIT_ASSERT_EQUAL(NPersQueue::NErrorCode::OVERLOAD, event->ErrorCode);
}
TestWriteSubDomainOutOfSpace_DeadlineWork(false);
}

Y_UNIT_TEST_F(WriteSubDomainOutOfSpace_DisableExpiration, TPartitionFixture)
Expand All @@ -2384,7 +2390,7 @@ Y_UNIT_TEST_F(WriteSubDomainOutOfSpace_DisableExpiration, TPartitionFixture)

Y_UNIT_TEST_F(WriteSubDomainOutOfSpace_IgnoreQuotaDeadline, TPartitionFixture)
{
TestWriteSubDomainOutOfSpace(TDuration::MilliSeconds(300), true);
TestWriteSubDomainOutOfSpace_DeadlineWork(true);
}

Y_UNIT_TEST_F(GetPartitionWriteInfoSuccess, TPartitionFixture) {
Expand Down
Loading