Topic SDK direct read (#18263)

Author: qyryq

.github/last_commit.txt

@@ -1 +1 @@
-984787760ea6d7f0bbf954dc56d979d45d3aaf61
+3c6c01111a028a661c25d7ad10b9c81c755f334b

include/ydb-cpp-sdk/client/topic/control_plane.h

@@ -128,6 +128,7 @@ class TPartitionConsumerStats {
 class TPartitionLocation {
 public:
     TPartitionLocation(const Ydb::Topic::PartitionLocation& partitionLocation);
+    TPartitionLocation(std::int32_t nodeId, std::int64_t generation);
     int32_t GetNodeId() const;
     int64_t GetGeneration() const;
 

include/ydb-cpp-sdk/client/topic/read_events.h

@@ -1,10 +1,12 @@
 #pragma once
 
 #include "codecs.h"
+#include "control_plane.h"
 #include "events_common.h"
 
 #include <util/datetime/base.h>
 
+
 namespace NYdb::inline V3::NTopic {
 
 //! Partition session.
@@ -42,10 +44,14 @@ struct TPartitionSession: public TThrRefBase, public TPrintable<TPartitionSessio
     }
 
 protected:
+
     uint64_t PartitionSessionId;
     std::string TopicPath;
     std::string ReadSessionId;
     uint64_t PartitionId;
+    std::optional<TPartitionLocation> Location;
+    /*TDirectReadId*/ std::int64_t NextDirectReadId = 1;
+    std::optional</*TDirectReadId*/ std::int64_t> LastDirectReadId;
 };
 
 template<>

include/ydb-cpp-sdk/client/topic/read_session.h

@@ -193,6 +193,9 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
     //! AutoPartitioningSupport.
     FLUENT_SETTING_DEFAULT(bool, AutoPartitioningSupport, false);
 
+    // TODO(qyryq) Uncomment when direct read is ready.
+    // FLUENT_SETTING_DEFAULT(bool, DirectRead, false);
+
     //! Log.
     FLUENT_SETTING_OPTIONAL(TLog, Log);
 };

src/client/persqueue_public/ut/read_session_ut.cpp

@@ -513,7 +513,6 @@ class TReadSessionImplTestSetup {
     std::shared_ptr<TCallbackContext<TSingleClusterReadSessionImpl>> CbContext;
     std::shared_ptr<TThreadPool> ThreadPool;
     ::IExecutor::TPtr DefaultExecutor;
-    std::shared_ptr<std::unordered_map<ECodec, THolder<NTopic::ICodec>>> ProvidedCodecs = std::make_shared<std::unordered_map<ECodec, THolder<NTopic::ICodec>>>();
 };
 
 class TReorderingExecutor : public ::IExecutor {
@@ -588,10 +587,6 @@ TReadSessionImplTestSetup::TReadSessionImplTestSetup() {
         .Counters(MakeIntrusive<NYdb::NPersQueue::TReaderCounters>(MakeIntrusive<::NMonitoring::TDynamicCounters>()));
 
     Log.SetFormatter(GetPrefixLogFormatter(""));
-
-    (*ProvidedCodecs)[ECodec::GZIP] = MakeHolder<NTopic::TGzipCodec>();
-    (*ProvidedCodecs)[ECodec::LZOP] = MakeHolder<NTopic::TUnsupportedCodec>();
-    (*ProvidedCodecs)[ECodec::ZSTD] = MakeHolder<NTopic::TZstdCodec>();
 }
 
 TReadSessionImplTestSetup::~TReadSessionImplTestSetup() noexcept(false) {

src/client/persqueue_public/ut/ut_utils/ut_utils.h

@@ -14,12 +14,22 @@ using NYdb::NTopic::IAsyncExecutor;
 
 namespace NYdb::NPersQueue::NTests {
 
+struct TPersQueueYdbSdkTestSetupSettings {
+    TString TestCaseName;
+    bool Start = true;
+    TVector<NKikimrServices::EServiceKikimr> LogServices = ::NPersQueue::TTestServer::LOGGED_SERVICES;
+    NActors::NLog::EPriority LogPriority = NActors::NLog::PRI_DEBUG;
+    ui32 NodeCount = NKikimr::NPersQueueTests::PQ_DEFAULT_NODE_COUNT;
+    size_t TopicPartitionsCount = 1;
+};
+
 class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
     THolder<NYdb::TDriver> Driver;
     THolder<NYdb::NPersQueue::TPersQueueClient> PersQueueClient;
 
     TAdaptiveLock Lock;
 public:
+    // TODO(qyryq) Delete this ctor in favor of TPersQueueYdbSdkTestSetupSettings.
     TPersQueueYdbSdkTestSetup(const TString& testCaseName, bool start = true,
                               const TVector<NKikimrServices::EServiceKikimr>& logServices = ::NPersQueue::TTestServer::LOGGED_SERVICES,
                               NActors::NLog::EPriority logPriority = NActors::NLog::PRI_DEBUG,
@@ -29,6 +39,17 @@ class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
     {
     }
 
+    TPersQueueYdbSdkTestSetup(TPersQueueYdbSdkTestSetupSettings settings)
+        : SDKTestSetup(
+            settings.TestCaseName,
+            settings.Start,
+            settings.LogServices,
+            settings.LogPriority,
+            settings.NodeCount,
+            settings.TopicPartitionsCount)
+    {
+    }
+
     ~TPersQueueYdbSdkTestSetup() {
         if (PersQueueClient) {
             PersQueueClient = nullptr;