more fixes

Author: nshestakov

ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp

@@ -205,7 +205,8 @@ class TTopicReader: public TBaseProxyActor<TTopicReader> {
             x->Confirm();
             return WaitEvent(ev->Get()->Sender, ev->Get()->Cookie);
         } else if (auto* x = std::get_if<TReadSessionEvent::TEndPartitionSessionEvent>(&*event)) {
-            // do nothing.
+            x->Confirm();
+            return WaitEvent(ev->Get()->Sender, ev->Get()->Cookie);
         } else if (auto* x = std::get_if<TReadSessionEvent::TDataReceivedEvent>(&*event)) {
             if (AutoCommit) {
                 DeferredCommit.Add(*x);

ydb/public/sdk/cpp/client/ydb_topic/impl/event_handlers.cpp

@@ -76,7 +76,8 @@ class TGracefulReleasingSimpleDataHandlers : public TThrRefBase {
         }
     }
 
-    void OnEndPartitionStream(TReadSessionEvent::TEndPartitionSessionEvent&) {
+    void OnEndPartitionStream(TReadSessionEvent::TEndPartitionSessionEvent& event) {
+        event.Confirm();
     }
 
     void OnPartitionStreamClosed(TReadSessionEvent::TPartitionSessionClosedEvent& event) {
@@ -139,7 +140,8 @@ TReadSessionSettings::TEventHandlers& TReadSessionSettings::TEventHandlers::Simp
         StopPartitionSessionHandler([](TReadSessionEvent::TStopPartitionSessionEvent& event) {
             event.Confirm();
         });
-        EndPartitionSessionHandler([](TReadSessionEvent::TEndPartitionSessionEvent&) {
+        EndPartitionSessionHandler([](TReadSessionEvent::TEndPartitionSessionEvent& event) {
+            event.Confirm();
         });
         CommitOffsetAcknowledgementHandler([](TReadSessionEvent::TCommitOffsetAcknowledgementEvent&){});
         PartitionSessionClosedHandler([](TReadSessionEvent::TPartitionSessionClosedEvent&){});

ydb/public/sdk/cpp/client/ydb_topic/impl/read_session_impl.ipp

@@ -482,7 +482,7 @@ inline void TSingleClusterReadSessionImpl<false>::InitImpl(TDeferredActions<fals
     auto& init = *req.mutable_init_request();
 
     init.set_consumer(Settings.ConsumerName_);
-    init.set_autoscaling_support(Settings.AutoscalingSupport_.GetOrElse(false));
+    init.set_autoscaling_support(Settings.AutoscalingSupport_);
 
     for (const TTopicReadSettings& topic : Settings.Topics_) {
         auto* topicSettings = init.add_topics_read_settings();
@@ -1888,7 +1888,7 @@ bool TSingleClusterReadSessionImpl<UseMigrationProtocol>::AllParentSessionsHasBe
 
 template<bool UseMigrationProtocol>
 void TSingleClusterReadSessionImpl<UseMigrationProtocol>::ConfirmPartitionStreamEnd(TPartitionStreamImpl<UseMigrationProtocol>* partitionStream, const std::vector<ui32>& childIds) {
-    ReadingFinishedData.insert(partitionStream->GetAssignId()); // Check
+    ReadingFinishedData.insert(partitionStream->GetPartitionSessionId());
     for (auto& [_, s] : PartitionStreams) {
         for (auto partitionId : childIds) {
             if (s->GetPartitionId() == partitionId) {
@@ -2145,12 +2145,12 @@ TReadSessionEventsQueue<UseMigrationProtocol>::GetEventImpl(size_t& maxByteSize,
             TParent::Events.pop();
 
             if constexpr (!UseMigrationProtocol) {
-                // if (std::holds_alternative<TReadSessionEvent::TEndPartitionSessionEvent>(*event)) {
-                //     auto& e = std::get<TReadSessionEvent::TEndPartitionSessionEvent>(*event);
-                //     if (auto session = frontCbContext->LockShared()) {
-                //         session->SetReadingFinished(partitionStream->GetPartitionSessionId(), e.GetChildPartitionIds());
-                //     }
-                // }
+                if (std::holds_alternative<TReadSessionEvent::TPartitionSessionClosedEvent>(*event)) {
+                     auto& e = std::get<TReadSessionEvent::TPartitionSessionClosedEvent>(*event);
+                     if (auto session = frontCbContext->LockShared()) {
+                        session->UnregisterPartition(e.GetPartitionSession()->GetPartitionId(), e.GetPartitionSession()->GetPartitionSessionId());
+                     }
+                }
             }
         }
 

ydb/public/sdk/cpp/client/ydb_topic/include/read_session.h

@@ -194,7 +194,7 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
     FLUENT_SETTING_DEFAULT(TDuration, ConnectTimeout, TDuration::Seconds(30));
 
     //! AutoscalingSupport.
-    FLUENT_SETTING_OPTIONAL(bool, AutoscalingSupport);
+    FLUENT_SETTING_DEFAULT(bool, AutoscalingSupport, false);
 
     //! Log.
     FLUENT_SETTING_OPTIONAL(TLog, Log);

ydb/public/sdk/cpp/client/ydb_topic/ut/basic_usage_ut.cpp

@@ -794,8 +794,8 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
                 [&](TReadSessionEvent::TStopPartitionSessionEvent& event) {
                     event.Confirm();
                 },
-                [&](TReadSessionEvent::TEndPartitionSessionEvent&) {
-                    // do nothing
+                [&](TReadSessionEvent::TEndPartitionSessionEvent& event) {
+                    event.Confirm();
                 },
                 [&](TReadSessionEvent::TPartitionSessionStatusEvent&) {
                     UNIT_FAIL("Test does not support lock sessions yet");

ydb/public/sdk/cpp/examples/topic_reader/eventloop/main.cpp

@@ -104,8 +104,8 @@ int main(int argc, const char* argv[]) {
             startPartitionSessionEvent->Confirm();
         } else if (auto* stopPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TStopPartitionSessionEvent>(&*event)) {
             stopPartitionSessionEvent->Confirm();
-        } else if (auto* stopPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TEndPartitionSessionEvent>(&*event)) {
-            // commit messages
+        } else if (auto* endPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TEndPartitionSessionEvent>(&*event)) {
+            endPartitionSessionEvent->Confirm();
         } else if (auto* closeSessionEvent = std::get_if<NYdb::NTopic::TSessionClosedEvent>(&*event)) {
             break;
         }