Merge 247e41c into 261a36a

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