Merge pull request #18017 from nshestakov/Kafka-improvements-25-1

Kafka improvements

Author: nshestakov

ydb/core/grpc_services/rpc_login.cpp

@@ -105,10 +105,15 @@ class TLoginRPC : public TRpcRequestActor<TLoginRPC, TEvLoginRequest, true> {
         }
     }
 
+    void HandleDestroyed(TEvTabletPipe::TEvClientDestroyed::TPtr&) {
+        ReplyErrorAndPassAway(Ydb::StatusIds::UNAVAILABLE, "SchemeShard is unavailable");
+    }
+
     STATEFN(StateWork) {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvents::TEvUndelivered, HandleUndelivered);
             hFunc(TEvTabletPipe::TEvClientConnected, HandleConnect);
+            hFunc(TEvTabletPipe::TEvClientDestroyed, HandleDestroyed);
             hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleNavigate);
             hFunc(TEvSchemeShard::TEvLoginResult, HandleResult);
             hFunc(TEvLdapAuthProvider::TEvAuthenticateResponse, Handle);

ydb/core/kafka_proxy/actors/actors.h

@@ -52,9 +52,9 @@ struct TContext {
     NKikimr::NPQ::TRlContext RlContext;
 
     bool Authenticated() {
-        
+
         return !RequireAuthentication || AuthenticationStep == SUCCESS;
-    
+
     }
 
     TActorId DiscoveryCacheActor;
@@ -176,6 +176,7 @@ NActors::IActor* CreateKafkaMetadataActor(const TContext::TPtr context, const ui
                                           const TActorId& discoveryCacheActor);
 NActors::IActor* CreateKafkaProduceActor(const TContext::TPtr context);
 NActors::IActor* CreateKafkaReadSessionActor(const TContext::TPtr context, ui64 cookie);
+NActors::IActor* CreateKafkaBalancerActor(const TContext::TPtr context, ui64 cookie);
 NActors::IActor* CreateKafkaSaslHandshakeActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TSaslHandshakeRequestData>& message);
 NActors::IActor* CreateKafkaSaslAuthActor(const TContext::TPtr context, const ui64 correlationId, const NKikimr::NRawSocket::TSocketDescriptor::TSocketAddressType address, const TMessagePtr<TSaslAuthenticateRequestData>& message);
 NActors::IActor* CreateKafkaListOffsetsActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TListOffsetsRequestData>& message);
@@ -185,6 +186,7 @@ NActors::IActor* CreateKafkaOffsetCommitActor(const TContext::TPtr context, cons
 NActors::IActor* CreateKafkaOffsetFetchActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TOffsetFetchRequestData>& message);
 NActors::IActor* CreateKafkaCreateTopicsActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TCreateTopicsRequestData>& message);
 NActors::IActor* CreateKafkaCreatePartitionsActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TCreatePartitionsRequestData>& message);
+NActors::IActor* CreateKafkaDescribeConfigsActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TDescribeConfigsRequestData>& message);
 NActors::IActor* CreateKafkaAlterConfigsActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TAlterConfigsRequestData>& message);
 
 } // namespace NKafka

ydb/core/kafka_proxy/actors/control_plane_common.h

@@ -63,13 +63,13 @@ inline TRetentionsConversionResult ConvertRetentions(std::optional<TString> rete
             RETENTION_MS_CONFIG_NAME,
             [&result](std::optional<ui64> retention) -> void { result.Ms = retention; }
     );
-    
+
     convertRetention(
             retentionBytes,
             RETENTION_BYTES_CONFIG_NAME,
             [&result](std::optional<ui64> retention) -> void { result.Bytes = retention; }
     );
-    
+
     return result;
 }
 
@@ -107,7 +107,7 @@ inline std::optional<THolder<TEvKafka::TEvTopicModificationResponse>> ValidateTo
     } else {
         return std::optional<THolder<TEvKafka::TEvTopicModificationResponse>>();
     }
-} 
+}
 
 template<class T>
 inline std::unordered_set<TString> ExtractDuplicates(
@@ -134,15 +134,15 @@ class TAlterTopicActor : public NKikimr::NGRpcProxy::V1::TUpdateSchemeActor<T, U
 public:
 
     TAlterTopicActor(
-            TActorId requester, 
+            TActorId requester,
             TIntrusiveConstPtr<NACLib::TUserToken> userToken,
             TString topicPath,
             TString databaseName)
         : TBase(new U(
             userToken,
             topicPath,
             databaseName,
-            [this](const EKafkaErrors status, const std::string& message) {
+            [this](const EKafkaErrors status, const std::string& message, const ::google::protobuf::Message&) {
                 this->SendResult(status,TString{message});
             })
         )
@@ -176,15 +176,15 @@ class TAlterTopicActor : public NKikimr::NGRpcProxy::V1::TUpdateSchemeActor<T, U
     const std::shared_ptr<TString> SerializedToken;
 };
 
-class TKafkaTopicModificationRequest : public NKikimr::NGRpcService::IRequestOpCtx {
+class TKafkaTopicRequestCtx : public NKikimr::NGRpcService::IRequestOpCtx {
 public:
-    using TRequest = TKafkaTopicModificationRequest;
+    using TRequest = TKafkaTopicRequestCtx;
 
-    TKafkaTopicModificationRequest(
+    TKafkaTopicRequestCtx(
             TIntrusiveConstPtr<NACLib::TUserToken> userToken,
             TString topicPath,
             TString databaseName,
-            const std::function<void(const EKafkaErrors, const std::string&)> sendResultCallback)
+            const std::function<void(const EKafkaErrors, const std::string&, const google::protobuf::Message& result)> sendResultCallback)
         : UserToken(userToken)
         , TopicPath(topicPath)
         , DatabaseName(databaseName)
@@ -239,7 +239,7 @@ class TKafkaTopicModificationRequest : public NKikimr::NGRpcService::IRequestOpC
     };
 
     void ReplyWithYdbStatus(Ydb::StatusIds::StatusCode status) override {
-        ProcessYdbStatusCode(status);
+        ProcessYdbStatusCode(status, google::protobuf::Empty{});
     };
 
     void ReplyWithRpcStatus(grpc::StatusCode code, const TString& msg = "", const TString& details = "") override {
@@ -334,26 +334,24 @@ class TKafkaTopicModificationRequest : public NKikimr::NGRpcService::IRequestOpC
     }
 
     void SendResult(const google::protobuf::Message& result, Ydb::StatusIds::StatusCode status) override {
-        Y_UNUSED(result);
-        ProcessYdbStatusCode(status);
+        ProcessYdbStatusCode(status, result);
     };
 
     void SendResult(
             const google::protobuf::Message& result,
             Ydb::StatusIds::StatusCode status,
             const google::protobuf::RepeatedPtrField<NKikimr::NGRpcService::TYdbIssueMessageType>& message) override {
 
-        Y_UNUSED(result);
         Y_UNUSED(message);
-        ProcessYdbStatusCode(status);
+        ProcessYdbStatusCode(status, result);
     };
 
     const Ydb::Operations::OperationParams& operation_params() const {
         return DummyParams;
     }
 
-    static TKafkaTopicModificationRequest* GetProtoRequest(std::shared_ptr<IRequestOpCtx> request) {
-        return static_cast<TKafkaTopicModificationRequest*>(&(*request));
+    static TKafkaTopicRequestCtx* GetProtoRequest(std::shared_ptr<IRequestOpCtx> request) {
+        return static_cast<TKafkaTopicRequestCtx*>(&(*request));
     }
 
 protected:
@@ -371,11 +369,12 @@ class TKafkaTopicModificationRequest : public NKikimr::NGRpcService::IRequestOpC
     const NKikimr::NGRpcService::TAuditLogParts DummyAuditLogParts;
     const TString TopicPath;
     const TString DatabaseName;
-    const std::function<void(const EKafkaErrors status, const std::string& message)> SendResultCallback;
+    const std::function<void(const EKafkaErrors status, const std::string& message, const google::protobuf::Message& result)> SendResultCallback;
     NYql::TIssue Issue;
 
-    void ProcessYdbStatusCode(Ydb::StatusIds::StatusCode& status) {
-        SendResultCallback(Convert(status), Issue.GetMessage());
+    void ProcessYdbStatusCode(Ydb::StatusIds::StatusCode& status, const google::protobuf::Message& result) {
+        SendResultCallback(Convert(status), Issue.GetMessage(), result);
     }
 };
-}
+
+} //namespace NKafka

ydb/core/kafka_proxy/actors/kafka_alter_configs_actor.cpp

@@ -11,22 +11,22 @@
 
 namespace NKafka {
 
-class TKafkaAlterConfigsRequest: public TKafkaTopicModificationRequest {
+class TKafkaAlterConfigsRequest: public TKafkaTopicRequestCtx {
 public:
     TKafkaAlterConfigsRequest(
             TIntrusiveConstPtr<NACLib::TUserToken> userToken,
             TString topicPath,
             TString databaseName,
-            const std::function<void(const EKafkaErrors, const std::string&)> sendResultCallback)
-        : TKafkaTopicModificationRequest(userToken, topicPath, databaseName, sendResultCallback)
+            const std::function<void(const EKafkaErrors, const std::string&, const google::protobuf::Message&)> sendResultCallback)
+        : TKafkaTopicRequestCtx(userToken, topicPath, databaseName, sendResultCallback)
     {
     };
 
 protected:
     EKafkaErrors Convert(Ydb::StatusIds::StatusCode& status) override {
         return status == Ydb::StatusIds::BAD_REQUEST
                 ? INVALID_CONFIG
-                : TKafkaTopicModificationRequest::Convert(status);
+                : TKafkaTopicRequestCtx::Convert(status);
     }
 };