Skip to content

Fix the processing of incorrect SourceId #1555

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 2 commits into from
Feb 5, 2024
Merged

Fix the processing of incorrect SourceId #1555

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c37ee9f
Fix the processing of incorrect SourceId
nshestakov Feb 2, 2024
0bbe4a2
fix
nshestakov Feb 5, 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
25 changes: 25 additions & 0 deletions ydb/core/persqueue/ut/partition_chooser_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -621,6 +621,19 @@ Y_UNIT_TEST(TPartitionChooserActor_SplitMergeEnabled_PreferedPartition_OtherPart
AssertTable(server, "A_Source_10", 0, 13);
}

Y_UNIT_TEST(TPartitionChooserActor_SplitMergeEnabled_BadSourceId_Test) {
NPersQueue::TTestServer server = CreateServer();

auto config = CreateConfig0(true);
AddPartition(config, 0, {}, {});
CreatePQTabletMock(server, 0, ETopicPartitionStatus::Active);

auto r = ChoosePartition(server, config, "base64:a***");

UNIT_ASSERT(r->Error);
}


Y_UNIT_TEST(TPartitionChooserActor_SplitMergeDisabled_NewSourceId_Test) {
NPersQueue::TTestServer server = CreateServer();

Expand Down Expand Up @@ -692,4 +705,16 @@ Y_UNIT_TEST(TPartitionChooserActor_SplitMergeDisabled_PreferedPartition_Inactive
UNIT_ASSERT(r->Error);
}

Y_UNIT_TEST(TPartitionChooserActor_SplitMergeDisabled_BadSourceId_Test) {
NPersQueue::TTestServer server = CreateServer();

auto config = CreateConfig0(false);
AddPartition(config, 0, {}, {});

auto r = ChoosePartition(server, config, "base64:a***");

UNIT_ASSERT(r->Error);
}


}
9 changes: 7 additions & 2 deletions ydb/core/persqueue/writer/partition_chooser_impl__abstract_chooser_actor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,13 @@ class TAbstractPartitionChooserActor: public TActorBootstrapped<TDerived> {
return TActor<TDerived>::SelfId();
}

void Initialize(const NActors::TActorContext& ctx) {
TableHelper.Initialize(ctx, SourceId);
[[nodiscard]] bool Initialize(const NActors::TActorContext& ctx) {
if (TableHelper.Initialize(ctx, SourceId)) {
return true;
}
StartIdle();
TThis::ReplyError(ErrorCode::BAD_REQUEST, "Bad SourceId", ctx);
return false;
}

void PassAway() {
Expand Down
4 changes: 3 additions & 1 deletion ydb/core/persqueue/writer/partition_chooser_impl__old_chooser_actor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,9 @@ class TPartitionChooserActor: public TAbstractPartitionChooserActor<TPartitionCh
}

void Bootstrap(const TActorContext& ctx) {
TThis::Initialize(ctx);
if (!TThis::Initialize(ctx)) {
return;
}
TThis::InitTable(ctx);
}

Expand Down
4 changes: 3 additions & 1 deletion ydb/core/persqueue/writer/partition_chooser_impl__sm_chooser_actor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@ class TSMPartitionChooserActor: public TAbstractPartitionChooserActor<TSMPartiti
}

void Bootstrap(const TActorContext& ctx) {
TThis::Initialize(ctx);
if (!TThis::Initialize(ctx)) {
return;
}
BoundaryPartition = ChoosePartitionSync();

if (TThis::SourceId) {
Expand Down
2 changes: 1 addition & 1 deletion ydb/core/persqueue/writer/partition_chooser_impl__table_helper.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class TTableHelper {
return SeqNo_;
}

bool Initialize(const TActorContext& ctx, const TString& sourceId) {
[[nodiscard]] bool Initialize(const TActorContext& ctx, const TString& sourceId) {
const auto& pqConfig = AppData(ctx)->PQConfig;

TableGeneration = pqConfig.GetTopicsAreFirstClassCitizen() ? ESourceIdTableGeneration::PartitionMapping
Expand Down