summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--ydb/core/kafka_proxy/actors/kafka_offset_fetch_actor.cpp5
-rw-r--r--ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp9
-rw-r--r--ydb/core/kafka_proxy/actors/kafka_topic_offsets_actor.h8
-rw-r--r--ydb/core/kafka_proxy/ut/ut_protocol.cpp108
-rw-r--r--ydb/core/persqueue/fetch_request_actor.cpp59
-rw-r--r--ydb/public/lib/ydb_cli/commands/ydb_service_scheme.cpp4
-rw-r--r--ydb/services/lib/actors/pq_schema_actor.h26
-rw-r--r--ydb/services/persqueue_v1/actors/schema_actors.h3
8 files changed, 198 insertions, 24 deletions
diff --git a/ydb/core/kafka_proxy/actors/kafka_offset_fetch_actor.cpp b/ydb/core/kafka_proxy/actors/kafka_offset_fetch_actor.cpp
index 96893aaa59c..7f494d25e1b 100644
--- a/ydb/core/kafka_proxy/actors/kafka_offset_fetch_actor.cpp
+++ b/ydb/core/kafka_proxy/actors/kafka_offset_fetch_actor.cpp
@@ -28,7 +28,8 @@ struct PartitionOffsets {
class TTopicOffsetActor: public NKikimr::NGRpcProxy::V1::TPQInternalSchemaActor<TTopicOffsetActor,
NKikimr::NGRpcProxy::V1::TLocalRequestBase,
NKikimr::NGRpcProxy::V1::TEvPQProxy::TEvPartitionLocationResponse>,
- public NKikimr::NGRpcProxy::V1::TDescribeTopicActorImpl {
+ public NKikimr::NGRpcProxy::V1::TDescribeTopicActorImpl,
+ public NKikimr::NGRpcProxy::V1::TCdcStreamCompatible {
using TBase = NKikimr::NGRpcProxy::V1::TPQInternalSchemaActor<TTopicOffsetActor,
NKikimr::NGRpcProxy::V1::TLocalRequestBase,
NKikimr::NGRpcProxy::V1::TEvPQProxy::TEvPartitionLocationResponse>;
@@ -36,7 +37,7 @@ class TTopicOffsetActor: public NKikimr::NGRpcProxy::V1::TPQInternalSchemaActor<
public:
TTopicOffsetActor(std::shared_ptr<TSet<TString>> consumers,
const NKikimr::NGRpcProxy::V1::TLocalRequestBase& request,
- const TActorId& requester,
+ const TActorId& requester,
std::shared_ptr<TSet<ui32>> partitions,
const TString& originalTopicName,
const TString& userSID)
diff --git a/ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp b/ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp
index c770736b1f2..7cdcc4a5c44 100644
--- a/ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp
+++ b/ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp
@@ -134,12 +134,13 @@ bool TKafkaSaslAuthActor::TryParseAuthDataTo(TKafkaSaslAuthActor::TAuthData& aut
auto password = tokens[2];
size_t atPos = userAndDatabase.rfind('@');
if (atPos == TString::npos) {
- SendResponseAndDie(EKafkaErrors::SASL_AUTHENTICATION_FAILED, "Database not provided.", "", ctx);
- return false;
+ authData.UserName = "";
+ authData.Database = userAndDatabase;
+ } else {
+ authData.UserName = userAndDatabase.substr(0, atPos);
+ authData.Database = userAndDatabase.substr(atPos + 1);
}
- authData.UserName = userAndDatabase.substr(0, atPos);
- authData.Database = userAndDatabase.substr(atPos + 1);
authData.Password = password;
return true;
}
diff --git a/ydb/core/kafka_proxy/actors/kafka_topic_offsets_actor.h b/ydb/core/kafka_proxy/actors/kafka_topic_offsets_actor.h
index 08db21ce59c..c284a9d3651 100644
--- a/ydb/core/kafka_proxy/actors/kafka_topic_offsets_actor.h
+++ b/ydb/core/kafka_proxy/actors/kafka_topic_offsets_actor.h
@@ -14,7 +14,8 @@ namespace NKafka {
class TTopicOffsetsActor : public NKikimr::NGRpcProxy::V1::TPQInternalSchemaActor<TTopicOffsetsActor,
TEvKafka::TGetOffsetsRequest,
TEvKafka::TEvTopicOffsetsResponse>
- , public NKikimr::NGRpcProxy::V1::TDescribeTopicActorImpl {
+ , public NKikimr::NGRpcProxy::V1::TDescribeTopicActorImpl
+ , public NKikimr::NGRpcProxy::V1::TCdcStreamCompatible {
using TBase = TPQInternalSchemaActor<TTopicOffsetsActor,
TEvKafka::TGetOffsetsRequest,
@@ -30,17 +31,18 @@ public:
void StateWork(TAutoPtr<IEventHandle>& ev);
void HandleCacheNavigateResponse(NKikimr::TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) override;
-
+
virtual void ApplyResponse(TTabletInfo&, NKikimr::TEvPersQueue::TEvReadSessionsInfoResponse::TPtr&,
const TActorContext&) override {
Y_ABORT();
}
+
bool ApplyResponse(NKikimr::TEvPersQueue::TEvGetPartitionsLocationResponse::TPtr&, const TActorContext&) override {
Y_ABORT();
}
void ApplyResponse(TTabletInfo& tabletInfo, NKikimr::TEvPersQueue::TEvStatusResponse::TPtr& ev, const TActorContext& ctx) override;
-
+
void Reply(const TActorContext&) override;
void RaiseError(const TString& error, const Ydb::PersQueue::ErrorCode::ErrorCode errorCode, const Ydb::StatusIds::StatusCode status, const TActorContext&) override;
diff --git a/ydb/core/kafka_proxy/ut/ut_protocol.cpp b/ydb/core/kafka_proxy/ut/ut_protocol.cpp
index add3425602f..28514aa7073 100644
--- a/ydb/core/kafka_proxy/ut/ut_protocol.cpp
+++ b/ydb/core/kafka_proxy/ut/ut_protocol.cpp
@@ -1000,6 +1000,11 @@ Y_UNIT_TEST_SUITE(KafkaProtocol) {
TString topicName = "/Root/topic-0-test";
TString shortTopicName = "topic-0-test";
TString notExistsTopicName = "/Root/not-exists";
+
+ TString tableName = "/Root/table-0-test";
+ TString feedName = "feed";
+ TString feedPath = tableName + "/" + feedName;
+
ui64 minActivePartitions = 10;
TString key = "record-key";
@@ -1207,6 +1212,60 @@ Y_UNIT_TEST_SUITE(KafkaProtocol) {
}
}
+ {
+ NYdb::NTable::TTableClient tableClient(*testServer.Driver);
+ tableClient.RetryOperationSync([&](TSession session)
+ {
+ NYdb::NTable::TTableBuilder builder;
+ builder.AddNonNullableColumn("key", NYdb::EPrimitiveType::Int64).SetPrimaryKeyColumn("key");
+ builder.AddNonNullableColumn("value", NYdb::EPrimitiveType::Int64);
+
+ auto createResult = session.CreateTable(tableName, builder.Build()).ExtractValueSync();
+ UNIT_ASSERT_VALUES_EQUAL(createResult.IsTransportError(), false);
+ Cerr << createResult.GetIssues().ToString() << "\n";
+ UNIT_ASSERT_VALUES_EQUAL(createResult.GetStatus(), EStatus::SUCCESS);
+
+ auto alterResult = session.AlterTable(tableName, NYdb::NTable::TAlterTableSettings()
+ .AppendAddChangefeeds(NYdb::NTable::TChangefeedDescription(feedName,
+ NYdb::NTable::EChangefeedMode::Updates,
+ NYdb::NTable::EChangefeedFormat::Json))
+ ).ExtractValueSync();
+ Cerr << alterResult.GetIssues().ToString() << "\n";
+ UNIT_ASSERT_VALUES_EQUAL(alterResult.IsTransportError(), false);
+ UNIT_ASSERT_VALUES_EQUAL(alterResult.GetStatus(), EStatus::SUCCESS);
+ return alterResult;
+ }
+ );
+
+ TValueBuilder rows;
+ rows.BeginList();
+ rows.AddListItem()
+ .BeginStruct()
+ .AddMember("key").Int64(1)
+ .AddMember("value").Int64(2)
+ .EndStruct();
+ rows.EndList();
+
+ auto upsertResult = tableClient.BulkUpsert(tableName, rows.Build()).GetValueSync();
+ UNIT_ASSERT_EQUAL(upsertResult.GetStatus(), EStatus::SUCCESS);
+ }
+
+ {
+ // Check CDC
+ std::vector<std::pair<TString, std::vector<i32>>> topics {{feedPath, {0}}};
+ auto msg = client.Fetch(topics);
+ UNIT_ASSERT_VALUES_EQUAL(msg->Responses.size(), 1);
+ UNIT_ASSERT_VALUES_EQUAL(msg->Responses[0].Partitions.size(), 1);
+ UNIT_ASSERT_VALUES_EQUAL(msg->Responses[0].Partitions[0].ErrorCode, static_cast<TKafkaInt16>(EKafkaErrors::NONE_ERROR));
+ UNIT_ASSERT_VALUES_EQUAL(msg->Responses[0].Partitions[0].Records.has_value(), true);
+ UNIT_ASSERT_VALUES_EQUAL(msg->Responses[0].Partitions[0].Records->Records.size(), 1);
+ auto record = msg->Responses[0].Partitions[0].Records->Records[0];
+
+ auto data = record.Value.value();
+ auto dataStr = TString(data.data(), data.size());
+ UNIT_ASSERT_VALUES_EQUAL(dataStr, "{\"update\":{\"value\":2},\"key\":[1]}");
+ }
+
} // Y_UNIT_TEST(FetchScenario)
Y_UNIT_TEST(BalanceScenario) {
@@ -2300,4 +2359,53 @@ Y_UNIT_TEST_SUITE(KafkaProtocol) {
Sleep(TDuration::Seconds(1));
}
+
+ Y_UNIT_TEST(LoginWithApiKeyWithoutAt) {
+ TInsecureTestServer testServer;
+
+ TString topicName = "/Root/topic-0-test";
+
+ NYdb::NTopic::TTopicClient pqClient(*testServer.Driver);
+ {
+ auto result =
+ pqClient
+ .CreateTopic(topicName,
+ NYdb::NTopic::TCreateTopicSettings()
+ .PartitioningSettings(10, 100)
+ .BeginAddConsumer("consumer-0").EndAddConsumer())
+ .ExtractValueSync();
+ UNIT_ASSERT_VALUES_EQUAL(result.IsTransportError(), false);
+ UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SUCCESS);
+ }
+
+ auto settings = NTopic::TReadSessionSettings()
+ .AppendTopics(NTopic::TTopicReadSettings(topicName))
+ .ConsumerName("consumer-0");
+ auto topicReader = pqClient.CreateReadSession(settings);
+
+ TTestClient client(testServer.Port);
+
+ {
+ auto msg = client.ApiVersions();
+
+ UNIT_ASSERT_VALUES_EQUAL(msg->ErrorCode, static_cast<TKafkaInt16>(EKafkaErrors::NONE_ERROR));
+ UNIT_ASSERT_VALUES_EQUAL(msg->ApiKeys.size(), 18u);
+ }
+
+ {
+ auto msg = client.SaslHandshake();
+
+ UNIT_ASSERT_VALUES_EQUAL(msg->ErrorCode, static_cast<TKafkaInt16>(EKafkaErrors::NONE_ERROR));
+ UNIT_ASSERT_VALUES_EQUAL(msg->Mechanisms.size(), 1u);
+ UNIT_ASSERT_VALUES_EQUAL(*msg->Mechanisms[0], "PLAIN");
+ }
+
+ {
+ auto msg = client.SaslAuthenticate("/Root", "ApiKey-value-valid");
+ Cerr << msg->ErrorMessage << "\n";
+ UNIT_ASSERT_VALUES_EQUAL(msg->ErrorCode, static_cast<TKafkaInt16>(EKafkaErrors::NONE_ERROR));
+ }
+
+ Sleep(TDuration::Seconds(1));
+ }
} // Y_UNIT_TEST_SUITE(KafkaProtocol)
diff --git a/ydb/core/persqueue/fetch_request_actor.cpp b/ydb/core/persqueue/fetch_request_actor.cpp
index 2435c1024c2..e53eb503bf5 100644
--- a/ydb/core/persqueue/fetch_request_actor.cpp
+++ b/ydb/core/persqueue/fetch_request_actor.cpp
@@ -93,6 +93,9 @@ private:
TActorId RequesterId;
ui64 PendingQuotaAmount;
+ std::unordered_map<TString, TString> PrivateTopicPathToCdcPath;
+ std::unordered_map<TString, TString> CdcPathToPrivateTopicPath;
+
public:
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
return NKikimrServices::TActivity::PQ_FETCH_REQUEST;
@@ -176,16 +179,24 @@ public:
schemeCacheRequest->DatabaseName = Settings.Database;
THashSet<TString> topicsRequested;
- for (const auto& part : Settings.Partitions) {
- auto ins = topicsRequested.insert(part.Topic).second;
- if (!ins)
- continue;
- auto split = NKikimr::SplitPath(part.Topic);
+
+ if (PrivateTopicPathToCdcPath.empty()) {
+ for (const auto& part : Settings.Partitions) {
+ topicsRequested.insert(part.Topic);
+ }
+ } else {
+ for (const auto& [key, value] : PrivateTopicPathToCdcPath) {
+ topicsRequested.insert(key);
+ }
+ }
+
+ for (const auto& topicName : topicsRequested) {
+ auto split = NKikimr::SplitPath(topicName);
TSchemeCacheNavigate::TEntry entry;
entry.Path.insert(entry.Path.end(), split.begin(), split.end());
entry.SyncVersion = true;
- entry.ShowPrivatePath = false;
+ entry.ShowPrivatePath = true;
entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
schemeCacheRequest->ResultSet.emplace_back(std::move(entry));
@@ -197,6 +208,7 @@ public:
void HandleSchemeCacheResponse(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
LOG_DEBUG_S(ctx, NKikimrServices::PQ_FETCH_REQUEST, "Handle SchemeCache response");
auto& result = ev->Get()->Request;
+ bool anyCdcTopicInRequest = false;
for (const auto& entry : result->ResultSet) {
auto path = CanonizePath(NKikimr::JoinPath(entry.Path));
switch (entry.Status) {
@@ -219,6 +231,16 @@ public:
), ctx
);
}
+ if (entry.Kind == NSchemeCache::TSchemeCacheNavigate::KindCdcStream) {
+ anyCdcTopicInRequest = true;
+ Y_ABORT_UNLESS(entry.ListNodeEntry->Children.size() == 1);
+ auto privateTopicPath = CanonizePath(JoinPath(ChildPath(NKikimr::SplitPath(path), entry.ListNodeEntry->Children.at(0).Name)));
+ PrivateTopicPathToCdcPath[privateTopicPath] = path;
+ CdcPathToPrivateTopicPath[path] = privateTopicPath;
+ TopicInfo[privateTopicPath] = TopicInfo[path];
+ TopicInfo.erase(path);
+ continue;
+ }
if (entry.Kind != TSchemeCacheNavigate::EKind::KindTopic) {
return SendReplyAndDie(
CreateErrorReply(
@@ -256,6 +278,12 @@ public:
topicInfo.BalancerTabletId = description.GetBalancerTabletID();
topicInfo.PQInfo = entry.PQGroupInfo;
}
+
+ if (anyCdcTopicInRequest) {
+ SendSchemeCacheRequest(ctx);
+ return;
+ }
+
for (auto& p: TopicInfo) {
ProcessMetadata(p.first, p.second, ctx);
}
@@ -393,8 +421,15 @@ public:
return SendReplyAndDie(std::move(Response), ctx);
}
Y_ABORT_UNLESS(FetchRequestReadsDone < Settings.Partitions.size());
- const auto& req = Settings.Partitions[FetchRequestReadsDone];
- const auto& topic = req.Topic;
+ auto& req = Settings.Partitions[FetchRequestReadsDone];
+
+ auto& topic = req.Topic;
+
+ auto cdcToPrivateIt = CdcPathToPrivateTopicPath.find(req.Topic);
+ if (cdcToPrivateIt != CdcPathToPrivateTopicPath.end()) {
+ topic = cdcToPrivateIt->second;
+ }
+
const auto& offset = req.Offset;
const auto& part = req.Partition;
const auto& maxBytes = req.MaxBytes;
@@ -462,7 +497,13 @@ public:
const auto& topic = req.Topic;
const auto& part = req.Partition;
- res->SetTopic(topic);
+ auto privateTopicToCdcIt = PrivateTopicPathToCdcPath.find(topic);
+ if (privateTopicToCdcIt == PrivateTopicPathToCdcPath.end()) {
+ res->SetTopic(topic);
+ } else {
+ res->SetTopic(PrivateTopicPathToCdcPath[topic]);
+ }
+
res->SetPartition(part);
auto read = res->MutableReadResult();
if (record.HasPartitionResponse() && record.GetPartitionResponse().HasCmdReadResult())
diff --git a/ydb/public/lib/ydb_cli/commands/ydb_service_scheme.cpp b/ydb/public/lib/ydb_cli/commands/ydb_service_scheme.cpp
index 2099c5f4ef9..eb657ece3fa 100644
--- a/ydb/public/lib/ydb_cli/commands/ydb_service_scheme.cpp
+++ b/ydb/public/lib/ydb_cli/commands/ydb_service_scheme.cpp
@@ -338,6 +338,8 @@ namespace {
Cout << Endl << "MeteringMode: " << (TStringBuilder() << topicDescription.GetMeteringMode());
if (!topicDescription.GetSupportedCodecs().empty()) {
Cout << Endl << "SupportedCodecs: " << FormatCodecs(topicDescription.GetSupportedCodecs()) << Endl;
+ } else {
+ Cout << Endl;
}
}
@@ -352,6 +354,8 @@ namespace {
Cout << Endl << "DownUtilizationPercent: " << topicDescription.GetPartitioningSettings().GetAutoPartitioningSettings().GetDownUtilizationPercent();
Cout << Endl << "UpUtilizationPercent: " << topicDescription.GetPartitioningSettings().GetAutoPartitioningSettings().GetUpUtilizationPercent();
Cout << Endl << "StabilizationWindowSeconds: " << topicDescription.GetPartitioningSettings().GetAutoPartitioningSettings().GetStabilizationWindow().Seconds() << Endl;
+ } else {
+ Cout << Endl;
}
}
}
diff --git a/ydb/services/lib/actors/pq_schema_actor.h b/ydb/services/lib/actors/pq_schema_actor.h
index 584f352d482..37795c0cf66 100644
--- a/ydb/services/lib/actors/pq_schema_actor.h
+++ b/ydb/services/lib/actors/pq_schema_actor.h
@@ -178,7 +178,6 @@ namespace NKikimr::NGRpcProxy::V1 {
if (ProcessCdc(response)) {
return;
}
-
AddIssue(
FillIssue(
TStringBuilder() << "path '" << path << "' is not compatible scheme object",
@@ -294,6 +293,7 @@ namespace NKikimr::NGRpcProxy::V1 {
const TMaybe<TString>& GetCdcStreamName() const {
return CdcStreamName;
}
+
void HandleCacheNavigateResponse(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
return static_cast<TDerived*>(this)->HandleCacheNavigateResponse(ev);
}
@@ -541,11 +541,15 @@ namespace NKikimr::NGRpcProxy::V1 {
virtual void HandleCacheNavigateResponse(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) = 0;
TString GetTopicPath() const override {
- return TBase::TopicPath;
+ auto path = TBase::TopicPath;
+ if (PrivateTopicName) {
+ path = JoinPath(ChildPath(NKikimr::SplitPath(path), *PrivateTopicName));
+ }
+ return path;
}
- void SendDescribeProposeRequest() {
- return TBase::SendDescribeProposeRequest(this->ActorContext(), false);
+ void SendDescribeProposeRequest(bool showPrivate = false) {
+ return TBase::SendDescribeProposeRequest(this->ActorContext(), showPrivate);
}
bool HandleCacheNavigateResponseBase(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
@@ -594,16 +598,28 @@ namespace NKikimr::NGRpcProxy::V1 {
return false;
}
+ bool ProcessCdc(const NSchemeCache::TSchemeCacheNavigate::TEntry& response) override {
+ if constexpr (THasCdcStreamCompatibility<TDerived>::Value) {
+ if (static_cast<TDerived*>(this)->IsCdcStreamCompatible()) {
+ Y_ABORT_UNLESS(response.ListNodeEntry->Children.size() == 1);
+ PrivateTopicName = response.ListNodeEntry->Children.at(0).Name;
+ SendDescribeProposeRequest(true);
+ return true;
+ }
+ }
+ return false;
+ }
+
private:
TRequest Request;
TActorId Requester;
- TMaybe<TString> PrivateTopicName;
protected:
THolder<TEvResponse> Response;
TIntrusiveConstPtr<NSchemeCache::TSchemeCacheNavigate::TPQGroupInfo> PQGroupInfo;
TIntrusiveConstPtr<NSchemeCache::TSchemeCacheNavigate::TDirEntryInfo> Self;
+ TMaybe<TString> PrivateTopicName;
};
}
diff --git a/ydb/services/persqueue_v1/actors/schema_actors.h b/ydb/services/persqueue_v1/actors/schema_actors.h
index b6c00f7cdb1..a5a78290cef 100644
--- a/ydb/services/persqueue_v1/actors/schema_actors.h
+++ b/ydb/services/persqueue_v1/actors/schema_actors.h
@@ -418,7 +418,8 @@ private:
class TPartitionsLocationActor : public TPQInternalSchemaActor<TPartitionsLocationActor,
TGetPartitionsLocationRequest,
TEvPQProxy::TEvPartitionLocationResponse>
- , public TDescribeTopicActorImpl {
+ , public TDescribeTopicActorImpl
+ , public TCdcStreamCompatible {
using TBase = TPQInternalSchemaActor<TPartitionsLocationActor, TGetPartitionsLocationRequest,
TEvPQProxy::TEvPartitionLocationResponse>;
generated by cgit v1.3 (git 2.53.0) at 2026-07-13 19:41:22 +0000