Support for autopartitioning topics for CLI (#10832)

Author: nshestakov

ydb/public/lib/ydb_cli/commands/ydb_service_topic.cpp

@@ -835,6 +835,7 @@ namespace {
 
     NTopic::TReadSessionSettings TCommandTopicRead::PrepareReadSessionSettings() {
         NTopic::TReadSessionSettings settings;
+        settings.AutoPartitioningSupport(true);
         settings.ConsumerName(Consumer_);
         // settings.ReadAll(); // TODO(shmel1k@): change to read only original?
         if (Timestamp_.Defined()) {

ydb/public/lib/ydb_cli/topic/topic_read.cpp

@@ -283,6 +283,16 @@ namespace NYdb::NConsoleClient {
         return EXIT_SUCCESS;
     }
 
+    int TTopicReader::HandleEndPartitionSessionEvent(NTopic::TReadSessionEvent::TEndPartitionSessionEvent *event) {
+        if (!HasSession(event->GetPartitionSession()->GetPartitionSessionId())) {
+            return EXIT_SUCCESS;
+        }
+
+        event->Confirm();
+
+        return EXIT_SUCCESS;
+    }
+
     int TTopicReader::HandleEvent(NYdb::NTopic::TReadSessionEvent::TEvent& event, IOutputStream& output) {
         if (auto* dataEvent = std::get_if<NTopic::TReadSessionEvent::TDataReceivedEvent>(&event)) {
             return HandleDataReceivedEvent(dataEvent, output);
@@ -292,10 +302,12 @@ namespace NYdb::NConsoleClient {
             return HandleCommitOffsetAcknowledgementEvent(commitEvent);
         } else if (auto* partitionStatusEvent = std::get_if<NTopic::TReadSessionEvent::TPartitionSessionStatusEvent>(&event)) {
             return HandlePartitionSessionStatusEvent(partitionStatusEvent);
-        } else if (auto* stopPartitionSessionEvent = std::get_if<NTopic::TReadSessionEvent::TStopPartitionSessionEvent>(&event)) { 
+        } else if (auto* stopPartitionSessionEvent = std::get_if<NTopic::TReadSessionEvent::TStopPartitionSessionEvent>(&event)) {
             return HandleStopPartitionSessionEvent(stopPartitionSessionEvent);
         } else if (auto* partitionSessionClosedEvent = std::get_if<NTopic::TReadSessionEvent::TPartitionSessionClosedEvent>(&event)) {
             return HandlePartitionSessionClosedEvent(partitionSessionClosedEvent);
+        } else if (auto* endPartitionSessionEvent = std::get_if<NTopic::TReadSessionEvent::TEndPartitionSessionEvent>(&event)) {
+            return HandleEndPartitionSessionEvent(endPartitionSessionEvent);
         } else if (auto* sessionClosedEvent = std::get_if<NTopic::TSessionClosedEvent>(&event)) {
             ThrowOnError(*sessionClosedEvent);
             return 1;
@@ -339,4 +351,4 @@ namespace NYdb::NConsoleClient {
         }
         return EXIT_SUCCESS;
     }
-} // namespace NYdb::NConsoleClient
+} // namespace NYdb::NConsoleClient

ydb/public/lib/ydb_cli/topic/topic_read.h

@@ -71,6 +71,7 @@ namespace NYdb::NConsoleClient {
         int HandlePartitionSessionStatusEvent(NTopic::TReadSessionEvent::TPartitionSessionStatusEvent*);
         int HandleStopPartitionSessionEvent(NTopic::TReadSessionEvent::TStopPartitionSessionEvent*);
         int HandlePartitionSessionClosedEvent(NTopic::TReadSessionEvent::TPartitionSessionClosedEvent*);
+        int HandleEndPartitionSessionEvent(NTopic::TReadSessionEvent::TEndPartitionSessionEvent*);
         int HandleDataReceivedEvent(NTopic::TReadSessionEvent::TDataReceivedEvent*, IOutputStream&);
         int HandleCommitOffsetAcknowledgementEvent(NTopic::TReadSessionEvent::TCommitOffsetAcknowledgementEvent*);
         int HandleEvent(NTopic::TReadSessionEvent::TEvent&, IOutputStream&);
@@ -104,4 +105,4 @@ namespace NYdb::NConsoleClient {
 
         THashMap<ui64, std::pair<NTopic::TPartitionSession::TPtr, EReadingStatus>> ActivePartitionSessions_;
     };
-} // namespace NYdb::NConsoleClient
+} // namespace NYdb::NConsoleClient