YDB C++ SDK Import 8

cmake/testing.cmake

@@ -130,6 +130,9 @@ function(add_ydb_test)
       WORKING_DIRECTORY
         ${YDB_TEST_WORKING_DIRECTORY}
     )
+    target_link_libraries(${YDB_TEST_NAME} PRIVATE
+      GTest::gtest_main 
+    )
   else()
     add_yunittest(
       NAME
@@ -146,6 +149,9 @@ function(add_ydb_test)
       WORKING_DIRECTORY
         ${YDB_TEST_WORKING_DIRECTORY}
     )
+    target_link_libraries(${YDB_TEST_NAME} PRIVATE
+      cpp-testing-unittest_main
+    )
   endif()
 
   set_yunittest_property(

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

@@ -111,6 +111,7 @@ class TPartitionConsumerStats {
     const TInstant& GetLastReadTime() const;
     const TDuration& GetMaxReadTimeLag() const;
     const TDuration& GetMaxWriteTimeLag() const;
+    const TDuration& GetMaxCommittedTimeLag() const;
 
 private:
     uint64_t CommittedOffset_;
@@ -120,6 +121,7 @@ class TPartitionConsumerStats {
     TInstant LastReadTime_;
     TDuration MaxReadTimeLag_;
     TDuration MaxWriteTimeLag_;
+    TDuration MaxCommittedTimeLag_;
 };
 
 // Topic partition location
@@ -766,6 +768,8 @@ struct TDescribePartitionSettings: public TOperationRequestSettings<TDescribePar
 };
 
 // Settings for commit offset request.
-struct TCommitOffsetSettings : public TOperationRequestSettings<TCommitOffsetSettings> {};
+struct TCommitOffsetSettings : public TOperationRequestSettings<TCommitOffsetSettings> {
+    FLUENT_SETTING_OPTIONAL(std::string, ReadSessionId);
+};
 
 }  // namespace NYdb::NTopic

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

@@ -31,6 +31,11 @@ struct TPartitionSession: public TThrRefBase, public TPrintable<TPartitionSessio
         return TopicPath;
     }
 
+    //! Read session id.
+    const std::string& GetReadSessionId() const {
+        return ReadSessionId;
+    }
+
     //! Partition id.
     uint64_t GetPartitionId() const {
         return PartitionId;
@@ -39,6 +44,7 @@ struct TPartitionSession: public TThrRefBase, public TPrintable<TPartitionSessio
 protected:
     uint64_t PartitionSessionId;
     std::string TopicPath;
+    std::string ReadSessionId;
     uint64_t PartitionId;
 };
 

src/api/protos/ydb_topic.proto

@@ -752,9 +752,10 @@ message CommitOffsetRequest {
     int64 partition_id = 3;
     // Path of consumer.
     string consumer = 4;
-
     // Processed offset.
     int64 offset = 5;
+    // Read session identifier from StreamRead RPC.
+    string read_session_id = 6;
 }
 
 // Commit offset response sent from server to client.
@@ -809,6 +810,8 @@ message Consumer {
         google.protobuf.Duration max_read_time_lag = 2;
         // Maximum of differences between write timestamp and create timestamp for all messages, read during last minute.
         google.protobuf.Duration max_write_time_lag = 3;
+        // The difference between the write timestamp of the last commited message and the current time.
+        google.protobuf.Duration max_committed_time_lag = 5;
         // Bytes read statistics.
         MultipleWindowsStat bytes_read = 4;
     }
@@ -1213,6 +1216,8 @@ message DescribeConsumerResult {
         google.protobuf.Duration max_read_time_lag = 6;
         // Maximum of differences between write timestamp and create timestamp for all messages, read during last minute.
         google.protobuf.Duration max_write_time_lag = 7;
+        // The difference between the write timestamp of the last commited message and the current time.
+        google.protobuf.Duration max_committed_time_lag = 13;
 
         // How much bytes were read during several windows statistics from this partition.
         MultipleWindowsStat bytes_read = 8;

src/client/topic/impl/read_session_impl.h

@@ -605,6 +605,7 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
     template <bool V = UseMigrationProtocol, class = std::enable_if_t<!V>>
     TPartitionStreamImpl(ui64 partitionStreamId,
                          std::string topicPath,
+                         std::string readSessionId,
                          i64 partitionId,
                          i64 assignId,
                          i64 readOffset,
@@ -617,6 +618,7 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
     {
         TAPartitionStream<false>::PartitionSessionId = partitionStreamId;
         TAPartitionStream<false>::TopicPath = std::move(topicPath);
+        TAPartitionStream<false>::ReadSessionId = std::move(readSessionId);
         TAPartitionStream<false>::PartitionId = static_cast<ui64>(partitionId);
         MaxCommittedOffset = static_cast<ui64>(readOffset);
     }
@@ -1333,6 +1335,7 @@ class TSingleClusterReadSessionImpl : public TEnableSelfContext<TSingleClusterRe
     const std::string Database;
     const std::string SessionId;
     const std::string ClusterName;
+    std::string ReadSessionId;
     TLog Log;
     ui64 NextPartitionStreamId;
     ui64 PartitionStreamIdStep;

src/client/topic/impl/read_session_impl.ipp

@@ -971,6 +971,7 @@ inline void TSingleClusterReadSessionImpl<true>::OnReadDoneImpl(
     LOG_LAZY(Log, TLOG_INFO, GetLogPrefix() << "Server session id: " << msg.session_id());
 
     RetryState = nullptr;
+    ReadSessionId = msg.session_id();
 
     // Successful init. Do nothing.
     ContinueReadingDataImpl();
@@ -1222,6 +1223,7 @@ inline void TSingleClusterReadSessionImpl<false>::OnReadDoneImpl(
     Y_UNUSED(deferred);
 
     RetryState = nullptr;
+    ReadSessionId = msg.session_id();
 
     LOG_LAZY(Log, TLOG_INFO, GetLogPrefix() << "Server session id: " << msg.session_id());
 
@@ -1321,8 +1323,12 @@ inline void TSingleClusterReadSessionImpl<false>::OnReadDoneImpl(
     Y_ABORT_UNLESS(Lock.IsLocked());
 
     auto partitionStream = MakeIntrusive<TPartitionStreamImpl<false>>(
-        NextPartitionStreamId, msg.partition_session().path(), msg.partition_session().partition_id(),
-        msg.partition_session().partition_session_id(), msg.committed_offset(),
+        NextPartitionStreamId,
+        msg.partition_session().path(),
+        ReadSessionId,
+        msg.partition_session().partition_id(),
+        msg.partition_session().partition_session_id(),
+        msg.committed_offset(),
         SelfContext);
     NextPartitionStreamId += PartitionStreamIdStep;
 

src/client/topic/impl/topic.cpp

@@ -376,6 +376,7 @@ TPartitionConsumerStats::TPartitionConsumerStats(const Ydb::Topic::DescribeConsu
     , LastReadTime_(TInstant::Seconds(partitionStats.last_read_time().seconds()))
     , MaxReadTimeLag_(TDuration::Seconds(partitionStats.max_read_time_lag().seconds()))
     , MaxWriteTimeLag_(TDuration::Seconds(partitionStats.max_write_time_lag().seconds()))
+    , MaxCommittedTimeLag_(TDuration::Seconds(partitionStats.max_committed_time_lag().seconds()))
 {}
 
 uint64_t TPartitionConsumerStats::GetCommittedOffset() const {
@@ -406,6 +407,10 @@ const TDuration& TPartitionConsumerStats::GetMaxWriteTimeLag() const {
     return MaxWriteTimeLag_;
 }
 
+const TDuration& TPartitionConsumerStats::GetMaxCommittedTimeLag() const {
+    return MaxCommittedTimeLag_;
+}
+
 TPartitionLocation::TPartitionLocation(const Ydb::Topic::PartitionLocation& partitionLocation)
     : NodeId_(partitionLocation.node_id())
     , Generation_(partitionLocation.generation())

src/client/topic/impl/topic_impl.h

@@ -282,7 +282,9 @@ class TTopicClient::TImpl : public TClientImplCommon<TTopicClient::TImpl> {
         request.set_partition_id(partitionId);
         request.set_consumer(TStringType{consumerName});
         request.set_offset(offset);
-
+        if (settings.ReadSessionId_) {
+            request.set_read_session_id(*settings.ReadSessionId_);
+        }
         return RunSimple<Ydb::Topic::V1::TopicService, Ydb::Topic::CommitOffsetRequest, Ydb::Topic::CommitOffsetResponse>(
             std::move(request),
             &Ydb::Topic::V1::TopicService::Stub::AsyncCommitOffset,

src/client/topic/ut/ut_utils/topic_sdk_test_setup.cpp

@@ -66,6 +66,27 @@ TTopicDescription TTopicSdkTestSetup::DescribeTopic(const TString& path)
     return status.GetTopicDescription();
 }
 
+TConsumerDescription TTopicSdkTestSetup::DescribeConsumer(const TString& path, const TString& consumer)
+{
+    TTopicClient client(MakeDriver());
+
+    TDescribeConsumerSettings settings;
+    settings.IncludeStats(true);
+    settings.IncludeLocation(true);
+
+    auto status = client.DescribeConsumer(path, consumer, settings).GetValueSync();
+    UNIT_ASSERT(status.IsSuccess());
+
+    return status.GetConsumerDescription();
+}
+
+TStatus TTopicSdkTestSetup::Commit(const TString& path, const TString& consumerName, size_t partitionId, size_t offset) {
+    TTopicClient client(MakeDriver());
+
+    return client.CommitOffset(path, partitionId, consumerName, offset).GetValueSync();
+}
+
+
 TString TTopicSdkTestSetup::GetEndpoint() const {
     return "localhost:" + ToString(Server.GrpcPort);
 }

src/client/topic/ut/ut_utils/topic_sdk_test_setup.h

@@ -22,6 +22,9 @@ class TTopicSdkTestSetup {
                      size_t maxPartitionCount = 100);
 
     TTopicDescription DescribeTopic(const TString& path = TString{TEST_TOPIC});
+    TConsumerDescription DescribeConsumer(const TString& path = TString{TEST_TOPIC}, const TString& consumer = TString{TEST_CONSUMER});
+
+    TStatus Commit(const TString& path, const TString& consumerName, size_t partitionId, size_t offset);
 
     TString GetEndpoint() const;
     TString GetTopicPath(const TString& name = TString{TEST_TOPIC}) const;

tests/integration/basic_example/CMakeLists.txt

@@ -1,4 +1,4 @@
-add_ydb_test(NAME basic-example
+add_ydb_test(NAME basic-example GTEST
   SOURCES
     main.cpp
     basic_example_data.cpp
@@ -9,7 +9,6 @@ add_ydb_test(NAME basic-example
     YDB-CPP-SDK::Driver
     YDB-CPP-SDK::Proto
     YDB-CPP-SDK::Table
-    GTest::gtest_main
   LABELS
     integration
 )

tests/integration/bulk_upsert/CMakeLists.txt

@@ -1,12 +1,11 @@
-add_ydb_test(NAME bulk_upsert
+add_ydb_test(NAME bulk_upsert GTEST
   SOURCES
     main.cpp
     bulk_upsert.cpp
     bulk_upsert.h
   LINK_LIBRARIES
     yutil
     YDB-CPP-SDK::Table
-    GTest::gtest_main
   LABELS
     integration
 )

tests/integration/server_restart/CMakeLists.txt

@@ -1,12 +1,11 @@
-add_ydb_test(NAME server_restart
+add_ydb_test(NAME server_restart GTEST
   SOURCES
     main.cpp
   LINK_LIBRARIES
     yutil
     api-grpc
     YDB-CPP-SDK::Query
     gRPC::grpc++
-    GTest::gtest_main
   LABELS
     integration
 )

tests/unit/client/CMakeLists.txt

@@ -3,7 +3,6 @@ add_ydb_test(NAME client-ydb_coordination_ut
     coordination/coordination_ut.cpp
   LINK_LIBRARIES
     yutil
-    cpp-testing-unittest_main
     YDB-CPP-SDK::Coordination
     api-grpc
   LABELS
@@ -15,7 +14,6 @@ add_ydb_test(NAME client-extensions-discovery_mutator_ut
     discovery_mutator/discovery_mutator_ut.cpp
   LINK_LIBRARIES
     yutil
-    cpp-testing-unittest_main
     YDB-CPP-SDK::DiscoveryMutator
     YDB-CPP-SDK::Table
   LABELS
@@ -27,7 +25,6 @@ add_ydb_test(NAME client-ydb_driver_ut
     driver/driver_ut.cpp
   LINK_LIBRARIES
     yutil
-    cpp-testing-unittest_main
     YDB-CPP-SDK::Driver
     YDB-CPP-SDK::Table
   LABELS
@@ -41,7 +38,6 @@ add_ydb_test(NAME client-impl-ydb_endpoints_ut
     endpoints/endpoints_ut.cpp
   LINK_LIBRARIES
     yutil
-    cpp-testing-unittest_main
     client-impl-ydb_endpoints
   LABELS
     unit
@@ -53,7 +49,6 @@ add_ydb_test(NAME client-oauth2_ut
     oauth2_token_exchange/jwt_token_source_ut.cpp
   LINK_LIBRARIES
     yutil
-    cpp-testing-unittest_main
     http-server
     json
     string_utils-base64
@@ -62,40 +57,34 @@ add_ydb_test(NAME client-oauth2_ut
     unit
 )
 
-# add_ydb_test(NAME client-ydb_params_ut
-#   SOURCES
-#     params/params_ut.cpp
-#   LINK_LIBRARIES
-#     yutil
-#     cpp-testing-unittest_main
-#     YDB-CPP-SDK::Params
-#     YDB-CPP-SDK::YsonValue
-#   LABELS
-#     unit
-# )
+add_ydb_test(NAME client-ydb_params_ut GTEST
+  SOURCES
+    params/params_ut.cpp
+  LINK_LIBRARIES
+    yutil
+    YDB-CPP-SDK::Params
+  LABELS
+    unit
+)
 
 add_ydb_test(NAME client-ydb_result_ut
   SOURCES
     result/result_ut.cpp
   LINK_LIBRARIES
     yutil
-    cpp-testing-unittest_main
     YDB-CPP-SDK::Result
     YDB-CPP-SDK::Params
   LABELS
     unit
 )
 
-# add_ydb_test(NAME client-ydb_value_ut
-#   SOURCES
-#     value/value_ut.cpp
-#   LINK_LIBRARIES
-#     yutil
-#     cpp-testing-unittest_main
-#     YDB-CPP-SDK::Value
-#     YDB-CPP-SDK::JsonValue
-#     YDB-CPP-SDK::YsonValue
-#     YDB-CPP-SDK::Params
-#   LABELS
-#     unit
-# )
+add_ydb_test(NAME client-ydb_value_ut GTEST
+  SOURCES
+    value/value_ut.cpp
+  LINK_LIBRARIES
+    yutil
+    YDB-CPP-SDK::Value
+    YDB-CPP-SDK::Params
+  LABELS
+    unit
+)