Fix kafka cdc read and login without @ (#10678)

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>;