Migrate logbroker to standalone YDB C++ SDK

Author: kobzonega

cloud/blockstore/libs/logbroker/iface/credentials_provider.cpp

@@ -2,7 +2,7 @@
 
 #include <cloud/blockstore/libs/logbroker/iface/config.h>
 
-#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/iam.h>
+#include <ydb-cpp-sdk/client/iam/iam.h>
 
 #include <util/string/cast.h>
 

cloud/blockstore/libs/logbroker/iface/ya.make

@@ -14,7 +14,7 @@ PEERDIR(
     library/cpp/monlib/service/pages
     library/cpp/threading/future
 
-    contrib/ydb/public/sdk/cpp/src/library/jwt
+    contrib/libs/ydb-cpp-sdk/src/client/iam
 )
 
 END()

cloud/blockstore/libs/logbroker/topic_api_impl/topic_api.cpp

@@ -7,9 +7,9 @@
 #include <cloud/storage/core/libs/common/error.h>
 #include <cloud/storage/core/libs/diagnostics/logging.h>
 
-#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/iam.h>
-#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/topic/client.h>
-#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/status_codes.h>
+#include <ydb-cpp-sdk/client/iam/iam.h>
+#include <ydb-cpp-sdk/client/topic/client.h>
+#include <ydb-cpp-sdk/client/types/status_codes.h>
 
 #include <util/generic/overloaded.h>
 #include <util/stream/file.h>
@@ -91,16 +91,19 @@ class TService final
 
     std::shared_ptr<NYdbICredentialsProviderFactory>
         CredentialsProviderFactory;
+    TWriteSessionFactory WriteSessionFactory;
 
 public:
     TService(
             TLogbrokerConfigPtr config,
             ILoggingServicePtr logging,
             std::shared_ptr<NYdbICredentialsProviderFactory>
-            credentialsProviderFactory)
+                credentialsProviderFactory,
+            TWriteSessionFactory writeSessionFactory)
         : Config(std::move(config))
         , Logging(std::move(logging))
         , CredentialsProviderFactory(std::move(credentialsProviderFactory))
+        , WriteSessionFactory(std::move(writeSessionFactory))
     {}
 
     TFuture<NProto::TError> Write(
@@ -119,15 +122,21 @@ class TService final
         Batch = std::make_unique<TBatch>(std::move(messages), now);
 
         if (!Session) {
-            TTopicClient client{GetDriver()};
-
-            Session = client.CreateWriteSession(
-                TWriteSessionSettings()
-                    .Path(Config->GetTopic())
-                    .ProducerId(Config->GetSourceId())
-                    .MessageGroupId(Config->GetSourceId())
-                    .RetryPolicy(
-                        NYdb::NTopic::IRetryPolicy::GetNoRetryPolicy()));
+            if (WriteSessionFactory) {
+                Session = WriteSessionFactory();
+            } else {
+                TTopicClient client{GetDriver()};
+
+                Session = client.CreateWriteSession(
+                    TWriteSessionSettings()
+                        .Path(Config->GetTopic())
+                        .ProducerId(Config->GetSourceId())
+                        .MessageGroupId(Config->GetSourceId())
+                        .RetryPolicy(
+                            NYdb::NTopic::IRetryPolicy::GetNoRetryPolicy()));
+            }
+
+            Y_ABORT_UNLESS(Session);
         } else if (ContinuationToken.has_value()) {
             TContinuationToken token{std::move(ContinuationToken.value())};
             ContinuationToken.reset();
@@ -314,13 +323,26 @@ class TService final
 IServicePtr CreateTopicAPIService(
     TLogbrokerConfigPtr config,
     ILoggingServicePtr logging,
-    std::shared_ptr<NYdbICredentialsProviderFactory>
-        credentialsProviderFactory)
+    std::shared_ptr<NYdbICredentialsProviderFactory> credentialsProviderFactory,
+    TWriteSessionFactory writeSessionFactory)
 {
     return std::make_shared<TService>(
         std::move(config),
         std::move(logging),
-        std::move(credentialsProviderFactory));
+        std::move(credentialsProviderFactory),
+        std::move(writeSessionFactory));
+}
+
+IServicePtr CreateTopicAPIService(
+    TLogbrokerConfigPtr config,
+    ILoggingServicePtr logging,
+    std::shared_ptr<NYdbICredentialsProviderFactory> credentialsProviderFactory)
+{
+    return CreateTopicAPIService(
+        std::move(config),
+        std::move(logging),
+        std::move(credentialsProviderFactory),
+        {});
 }
 
 IServicePtr CreateTopicAPIService(

cloud/blockstore/libs/logbroker/topic_api_impl/topic_api.h

@@ -5,10 +5,25 @@
 
 #include <cloud/storage/core/libs/diagnostics/public.h>
 
+#include <functional>
+
+namespace NYdb::inline V3::NTopic {
+class IWriteSession;
+}   // namespace NYdb::inline V3::NTopic
+
 namespace NCloud::NBlockStore::NLogbroker {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+using TWriteSessionFactory =
+    std::function<std::shared_ptr<NYdb::NTopic::IWriteSession>()>;
+
+IServicePtr CreateTopicAPIService(
+    TLogbrokerConfigPtr config,
+    ILoggingServicePtr logging,
+    std::shared_ptr<NYdbICredentialsProviderFactory> credentialsProviderFactory,
+    TWriteSessionFactory writeSessionFactory);
+
 IServicePtr CreateTopicAPIService(
     TLogbrokerConfigPtr config,
     ILoggingServicePtr logging,