Commit for autopartitioned topics (#11629)

Author: niksaveliev

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

@@ -766,6 +766,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.

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_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,