Moved "

Send EndPartitionSession control message" commit from ydb repo

Author: nshestakov

include/ydb-cpp-sdk/client/federated_topic/federated_topic.h

@@ -134,6 +134,7 @@ struct TReadSessionEvent {
     using TCommitOffsetAcknowledgementEvent = TFederated<NTopic::TReadSessionEvent::TCommitOffsetAcknowledgementEvent>;
     using TStartPartitionSessionEvent = TFederated<NTopic::TReadSessionEvent::TStartPartitionSessionEvent>;
     using TStopPartitionSessionEvent = TFederated<NTopic::TReadSessionEvent::TStopPartitionSessionEvent>;
+    using TEndPartitionSessionEvent = TFederated<NTopic::TReadSessionEvent::TEndPartitionSessionEvent>;
     using TPartitionSessionStatusEvent = TFederated<NTopic::TReadSessionEvent::TPartitionSessionStatusEvent>;
     using TPartitionSessionClosedEvent = TFederated<NTopic::TReadSessionEvent::TPartitionSessionClosedEvent>;
 
@@ -202,6 +203,7 @@ struct TReadSessionEvent {
                                 TCommitOffsetAcknowledgementEvent,
                                 TStartPartitionSessionEvent,
                                 TStopPartitionSessionEvent,
+                                TEndPartitionSessionEvent,
                                 TPartitionSessionStatusEvent,
                                 TPartitionSessionClosedEvent,
                                 TSessionClosedEvent>;
@@ -352,6 +354,13 @@ struct TFederatedReadSessionSettings: public NTopic::TReadSessionSettings {
         FLUENT_SETTING(std::function<void(TReadSessionEvent::TStopPartitionSessionEvent&)>,
                        StopPartitionSessionHandler);
 
+        //! Function to handle end partition session events.
+        //! If this handler is set, end partition session events will be handled by handler,
+        //! otherwise sent to TReadSession::GetEvent().
+        //! Default value is empty function (not set).
+        FLUENT_SETTING(std::function<void(TReadSessionEvent::TEndPartitionSessionEvent&)>,
+                       EndPartitionSessionHandler);
+
         //! Function to handle partition session status events.
         //! If this handler is set, partition session status events will be handled by handler,
         //! otherwise sent to TReadSession::GetEvent().
@@ -535,6 +544,8 @@ void TPrintable<NFederatedTopic::TReadSessionEvent::TFederated<NFederatedTopic::
 template<>
 void TPrintable<NFederatedTopic::TReadSessionEvent::TFederated<NFederatedTopic::TReadSessionEvent::TStopPartitionSessionEvent>>::DebugString(TStringBuilder& res, bool) const;
 template<>
+void TPrintable<NFederatedTopic::TReadSessionEvent::TFederated<NFederatedTopic::TReadSessionEvent::TEndPartitionSessionEvent>>::DebugString(TStringBuilder& res, bool) const;
+template<>
 void TPrintable<NFederatedTopic::TReadSessionEvent::TFederated<NFederatedTopic::TReadSessionEvent::TPartitionSessionStatusEvent>>::DebugString(TStringBuilder& res, bool) const;
 template<>
 void TPrintable<NFederatedTopic::TReadSessionEvent::TFederated<NFederatedTopic::TReadSessionEvent::TPartitionSessionClosedEvent>>::DebugString(TStringBuilder& res, bool) const;

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

@@ -282,6 +282,26 @@ struct TReadSessionEvent {
         ui64 CommittedOffset;
     };
 
+    //! Server command for ending partition session.
+    //! This is a hint that all messages from the partition have been read and will no longer appear, and that the client must commit offsets.
+    struct TEndPartitionSessionEvent: public TPartitionSessionAccessor, public TPrintable<TEndPartitionSessionEvent> {
+        TEndPartitionSessionEvent(TPartitionSession::TPtr partitionSession, std::vector<ui32>&& adjacentPartitionIds, std::vector<ui32>&& childPartitionIds);
+
+        //! A list of the partition IDs that also participated in the partition's merge.
+        const std::vector<ui32> GetAdjacentPartitionIds() const {
+            return AdjacentPartitionIds;
+        }
+
+        //! A list of partition IDs that were obtained as a result of merging or splitting this partition.
+        const std::vector<ui32> GetChildPartitionIds() const {
+            return ChildPartitionIds;
+        }
+
+    private:
+        std::vector<ui32> AdjacentPartitionIds;
+        std::vector<ui32> ChildPartitionIds;
+    };
+
     //! Status for partition session requested via TPartitionSession::RequestStatus()
     struct TPartitionSessionStatusEvent: public TPartitionSessionAccessor,
                                          public TPrintable<TPartitionSessionStatusEvent> {
@@ -343,6 +363,7 @@ struct TReadSessionEvent {
                                 TCommitOffsetAcknowledgementEvent,
                                 TStartPartitionSessionEvent,
                                 TStopPartitionSessionEvent,
+                                TEndPartitionSessionEvent,
                                 TPartitionSessionStatusEvent,
                                 TPartitionSessionClosedEvent,
                                 TSessionClosedEvent>;

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

@@ -104,6 +104,13 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
         FLUENT_SETTING(std::function<void(TReadSessionEvent::TStopPartitionSessionEvent&)>,
                        StopPartitionSessionHandler);
 
+        //! Function to handle end partition session events.
+        //! If this handler is set, end partition session events will be handled by handler,
+        //! otherwise sent to TReadSession::GetEvent().
+        //! Default value is empty function (not set).
+        FLUENT_SETTING(std::function<void(TReadSessionEvent::TEndPartitionSessionEvent&)>,
+                       EndPartitionSessionHandler);
+
         //! Function to handle partition session status events.
         //! If this handler is set, partition session status events will be handled by handler,
         //! otherwise sent to TReadSession::GetEvent().
@@ -186,6 +193,9 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
 
     FLUENT_SETTING_DEFAULT(TDuration, ConnectTimeout, TDuration::Seconds(30));
 
+    //! AutoscalingSupport.
+    FLUENT_SETTING_OPTIONAL(bool, AutoscalingSupport);
+
     //! Log.
     FLUENT_SETTING_OPTIONAL(TLog, Log);
 };

src/api/protos/ydb_persqueue_v1.proto

@@ -5,6 +5,8 @@ import "src/api/protos/ydb_status_codes.proto";
 import "src/api/protos/ydb_issue_message.proto";
 import "src/api/protos/annotations/validation.proto";
 
+import "google/protobuf/duration.proto";
+
 package Ydb.PersQueue.V1;
 
 option java_package = "com.yandex.ydb.persqueue";
@@ -1079,6 +1081,18 @@ message Credentials {
     }
 }
 
+enum AutoscalingStrategy {
+    // The autoscaling algorithm is not specified. The default value will be used.
+    AUTOSCALING_STRATEGY_UNSPECIFIED = 0;
+    // The autoscaling is disabled.
+    AUTOSCALING_STRATEGY_DISABLED = 1;
+    // The autoscaling algorithm will increase partitions count depending on the load characteristics.
+    // The autoscaling algorithm will never decrease the number of partitions.
+    AUTOSCALING_STRATEGY_SCALE_UP = 2;
+    // The autoscaling algorithm will both increase and decrease partitions count depending on the load characteristics.
+    AUTOSCALING_STRATEGY_SCALE_UP_AND_DOWN = 3;
+}
+
 /**
  * Message for describing topic internals.
  */
@@ -1088,8 +1102,13 @@ message TopicSettings {
         FORMAT_BASE = 1;
     }
 
-    // How many partitions in topic. Must less than database limit. Default limit - 10.
-    int32 partitions_count = 1 [(value) = "> 0"];
+    oneof partitioning {
+        // How many partitions in topic. Must less than database limit. Default limit - 10.
+        int32 partitions_count = 1 [(value) = "> 0"];
+        // Settings for the partitions count autoscaling.
+        AutoscalingSettings autoscaling_settings = 15;
+    };
+
     oneof retention {
         // How long data in partition should be stored. Must be greater than 0 and less than limit for this database.
         // Default limit - 36 hours.
@@ -1164,6 +1183,40 @@ message TopicSettings {
     RemoteMirrorRule remote_mirror_rule = 11;
 }
 
+message AutoscalingSettings {
+    // Strategy of autoscaling.
+    AutoscalingStrategy strategy = 1;
+
+    // Auto merge would stop working when the partitions count reaches min_active_partitions.
+    // Zero value means default - 1.
+    int64 min_active_partitions = 2 [(Ydb.value) = ">= 0"];
+    // Auto split would stop working when the partitions count reaches max_active_partitions.
+    // Zero value means default - 1.
+    int64 max_active_partitions = 3 [(Ydb.value) = ">= 0"];
+    // Limit for total partition count, including active (open for write) and read-only partitions.
+    // Zero value means default - 100.
+    int64 partition_count_limit = 4 [(Ydb.value) = ">= 0", deprecated = true];
+
+    // Partition write speed autoscaling options.
+    AutoscalingPartitionWriteSpeedStrategy partition_write_speed = 5;
+}
+
+message AutoscalingPartitionWriteSpeedStrategy {
+    //Partition will be autoscaled up (divided into 2 partitions)
+    //after write speed to the partition exceeds scale_up_threshold_percent (in percentage of maximum write speed to the partition) for the period of time threshold_time_seconds
+
+    //Partition will become a candidate to the autoscaling down
+    //after write speed doesn’t reach scale_down_threshold_percent (in percentage of maximum write speed to the partition) for the period of time threshold_time_seconds
+    //This candidate partition will be autoscaled down when other neighbour partition will become a candidate to the autoscaling down and not earlier than a retention period.
+
+    // Zero value means default - 300.
+    google.protobuf.Duration threshold_time = 1;
+    // Zero value means default - 90.
+    int32 scale_up_threshold_percent = 2 [(Ydb.value) = ">= 0"];
+    // Zero value means default - 30.
+    int32 scale_down_threshold_percent = 3 [(Ydb.value) = ">= 0"];
+}
+
 /**
  * Create topic request sent from client to server.
  */

src/api/protos/ydb_topic.proto

@@ -306,6 +306,7 @@ message StreamReadMessage {
             StopPartitionSessionRequest stop_partition_session_request = 9;
 
             UpdatePartitionSession update_partition_session = 10;
+            EndPartitionSession end_partition_session = 11;
         }
     }
 
@@ -320,6 +321,8 @@ message StreamReadMessage {
         string reader_name = 3;
         // Direct reading from a partition node.
         bool direct_read = 4;
+        // Indicates that the SDK supports autoscaling.
+        bool autoscaling_support = 5;
 
         message TopicReadSettings {
             // Topic path.
@@ -558,6 +561,20 @@ message StreamReadMessage {
         int64 direct_read_id = 2;
     }
 
+    // Signal from server that client has finished reading the partition and all messages have been read.
+    // Once a partition has been finished no further messages will ever arrive to that partition.
+    // This command is a hint to the client to commit offsets, after which the child partitions will be balanced independently in different reading sessions.
+    // Unlike StopPartitionSessionRequest, the client does not have to close the reading session.
+    // Client should not send a response to the command.
+    message EndPartitionSession {
+        // Partition session identifier.
+        int64 partition_session_id = 1;
+
+        // Ids of partitions which were merged with the ended partition.
+        repeated int64 adjacent_partition_ids = 2;
+        // Ids of partitions which was formed when the ended partition was split or merged.
+        repeated int64 child_partition_ids = 3;
+    }
 }
 
 // Messages for bidirectional streaming rpc StreamDirectRead
@@ -796,24 +813,93 @@ message AlterConsumer {
     map<string, string> alter_attributes = 6;
 }
 
+enum AutoscalingStrategy {
+    // The autoscaling algorithm is not specified. The default value will be used.
+    AUTOSCALING_STRATEGY_UNSPECIFIED = 0;
+    // The autoscaling is disabled.
+    AUTOSCALING_STRATEGY_DISABLED = 1;
+    // The autoscaling algorithm will increase partitions count depending on the load characteristics.
+    // The autoscaling algorithm will never decrease the number of partitions.
+    AUTOSCALING_STRATEGY_SCALE_UP = 2;
+    // The autoscaling algorithm will both increase and decrease partitions count depending on the load characteristics.
+    AUTOSCALING_STRATEGY_SCALE_UP_AND_DOWN = 3;
+}
+
 // Partitioning settings for topic.
 message PartitioningSettings {
-    // Minimum partition count auto merge would stop working at.
+    // Auto merge would stop working when the partitions count reaches min_active_partitions
     // Zero value means default - 1.
     int64 min_active_partitions = 1 [(Ydb.value) = ">= 0"];
+    // Auto split would stop working when the partitions count reaches max_active_partitions
+    // Zero value means default - 1.
+    int64 max_active_partitions = 3 [(Ydb.value) = ">= 0"];
     // Limit for total partition count, including active (open for write) and read-only partitions.
     // Zero value means default - 100.
-    int64 partition_count_limit = 2 [(Ydb.value) = ">= 0"];
+    // Use max_active_partitions
+    int64 partition_count_limit = 2 [(Ydb.value) = ">= 0", deprecated = true];
+    // Settings for the partitions count autoscaling.
+    AutoscalingSettings autoscaling_settings = 4;
+}
+
+message AutoscalingSettings {
+    // Strategy of autoscaling.
+    AutoscalingStrategy strategy = 1;
+    // Partition write speed autoscaling options.
+    AutoscalingPartitionWriteSpeedStrategy partition_write_speed = 2;
+}
+
+message AutoscalingPartitionWriteSpeedStrategy {
+    //Partition will be autoscaled up (divided into 2 partitions)
+    //after write speed to the partition exceeds scale_up_threshold_percent (in percentage of maximum write speed to the partition) for the period of time threshold_time_seconds
+
+    //Partition will become a candidate to the autoscaling down
+    //after write speed doesn’t reach scale_down_threshold_percent (in percentage of maximum write speed to the partition) for the period of time threshold_time_seconds
+    //This candidate partition will be autoscaled down when other neighbour partition will become a candidate to the autoscaling down and not earlier than a retention period.
+
+    // Zero value means default - 300.
+    google.protobuf.Duration threshold_time = 1;
+    // Zero value means default - 90.
+    int32 scale_up_threshold_percent = 2 [(Ydb.value) = ">= 0"];
+    // Zero value means default - 30.
+    int32 scale_down_threshold_percent = 3 [(Ydb.value) = ">= 0"];
 }
 
 // Partitioning settings for topic.
 message AlterPartitioningSettings {
     // Minimum partition count auto merge would stop working at.
     // Zero value means default - 1.
     optional int64 set_min_active_partitions = 1 [(Ydb.value) = ">= 0"];
+    // Maximum partition count auto merge would stop working at.
+    // Zero value means default - 1.
+    optional int64 max_active_partitions = 3 [(Ydb.value) = ">= 0"];
     // Limit for total partition count, including active (open for write) and read-only partitions.
     // Zero value means default - 100.
-    optional int64 set_partition_count_limit = 2 [(Ydb.value) = ">= 0"];
+    // Use set_max_active_partitions
+    optional int64 set_partition_count_limit = 2 [(Ydb.value) = ">= 0", deprecated = true];
+    // Settings for autoscaling the partition number
+    optional AlterAutoscalingSettings alter_autoscaling_settings = 4;
+}
+
+message AlterAutoscalingSettings {
+    // Strategy of autoscaling
+    optional AutoscalingStrategy set_strategy = 1;
+    // Autoscaling partition write speed options.
+    optional AlterAutoscalingPartitionWriteSpeedStrategy set_partition_write_speed = 2;
+}
+
+message AlterAutoscalingPartitionWriteSpeedStrategy {
+    // The time of exceeding the threshold value, after which the partition will be
+    // autoscaling.
+    // Zero value means default - 300.
+    optional google.protobuf.Duration set_threshold_time = 1;
+    // The threshold value of the write speed to the partition as a percentage, when exceeded,
+    // the partition will be auto split.
+    // Zero value means default - 90.
+    optional int32 set_scale_up_threshold_percent = 2 [(Ydb.value) = ">= 0"];
+    // The threshold value of the write speed to the partition as a percentage, if it is not reached,
+    // the partition will be auto merged.
+    // Zero value means default - 30.
+    optional int32 set_scale_down_threshold_percent = 3 [(Ydb.value) = ">= 0"];
 }
 
 // Metering mode specifies the method used to determine consumption of resources by the topic.

src/client/federated_topic/impl/federated_read_session.cpp

@@ -50,6 +50,7 @@ NTopic::TReadSessionSettings FromFederated(const TFederatedReadSessionSettings&
     MAYBE_CONVERT_HANDLER(TReadSessionEvent::TCommitOffsetAcknowledgementEvent, CommitOffsetAcknowledgementHandler);
     MAYBE_CONVERT_HANDLER(TReadSessionEvent::TStartPartitionSessionEvent, StartPartitionSessionHandler);
     MAYBE_CONVERT_HANDLER(TReadSessionEvent::TStopPartitionSessionEvent, StopPartitionSessionHandler);
+    MAYBE_CONVERT_HANDLER(TReadSessionEvent::TEndPartitionSessionEvent, EndPartitionSessionHandler);
     MAYBE_CONVERT_HANDLER(TReadSessionEvent::TPartitionSessionStatusEvent, PartitionSessionStatusHandler);
     MAYBE_CONVERT_HANDLER(TReadSessionEvent::TPartitionSessionClosedEvent, PartitionSessionClosedHandler);
     MAYBE_CONVERT_HANDLER(TReadSessionEvent::TEvent, CommonHandler);

src/client/federated_topic/impl/federated_read_session_event.cpp

@@ -11,6 +11,7 @@ using namespace NFederatedTopic;
 using TCommitOffsetAcknowledgementEvent = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TCommitOffsetAcknowledgementEvent>;
 using TStartPartitionSessionEvent = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TStartPartitionSessionEvent>;
 using TStopPartitionSessionEvent = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TStopPartitionSessionEvent>;
+using TEndPartitionSessionEvent = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TEndPartitionSessionEvent>;
 using TPartitionSessionStatusEvent = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TPartitionSessionStatusEvent>;
 using TPartitionSessionClosedEvent = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TPartitionSessionClosedEvent>;
 using TMessage = NFederatedTopic::TReadSessionEvent::TFederated<NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage>;
@@ -78,6 +79,20 @@ void TPrintable<TStopPartitionSessionEvent>::DebugString(TStringBuilder& ret, bo
         << " }";
 }
 
+void JoinIds(TStringBuilder& ret, const std::vector<ui32> ids);
+
+template<>
+void TPrintable<TEndPartitionSessionEvent>::DebugString(TStringBuilder& ret, bool) const {
+    const auto* self = static_cast<const TEndPartitionSessionEvent*>(this);
+    ret << "EndPartitionSession {";
+    self->GetFederatedPartitionSession()->DebugString(ret);
+    ret << " AdjacentPartitionIds: ";
+    JoinIds(ret, self->GetAdjacentPartitionIds());
+    ret << " ChildPartitionIds: ";
+    JoinIds(ret, self->GetChildPartitionIds());
+    ret << " }";
+}
+
 template<>
 void TPrintable<TPartitionSessionStatusEvent>::DebugString(TStringBuilder& ret, bool) const {
     const auto* self = static_cast<const TPartitionSessionStatusEvent*>(this);